Connector Development

Overview

Connectors are the integration layer between Release Orchestrator and external systems. Each connector implements a standard interface for its integration type.

Connector Architecture

                      CONNECTOR ARCHITECTURE

  ┌─────────────────────────────────────────────────────────────────────────────┐
  │                         CONNECTOR RUNTIME                                    │
  │                                                                             │
  │  ┌─────────────────────────────────────────────────────────────────────┐   │
  │  │                     CONNECTOR INTERFACE                              │   │
  │  │                                                                      │   │
  │  │  ┌──────────────────┐  ┌──────────────────┐  ┌──────────────────┐  │   │
  │  │  │ getCapabilities()│  │ ping()           │  │ authenticate()   │  │   │
  │  │  └──────────────────┘  └──────────────────┘  └──────────────────┘  │   │
  │  │                                                                      │   │
  │  │  ┌──────────────────┐  ┌──────────────────┐  ┌──────────────────┐  │   │
  │  │  │ discover()       │  │ execute()        │  │ healthCheck()    │  │   │
  │  │  └──────────────────┘  └──────────────────┘  └──────────────────┘  │   │
  │  │                                                                      │   │
  │  └─────────────────────────────────────────────────────────────────────┘   │
  │                                    │                                        │
  │                                    ▼                                        │
  │  ┌─────────────────────────────────────────────────────────────────────┐   │
  │  │                     CONNECTOR IMPLEMENTATIONS                        │   │
  │  │                                                                      │   │
  │  │  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐ │   │
  │  │  │ Registry    │  │ CI/CD       │  │ Notification│  │ Secret      │ │   │
  │  │  │ Connectors  │  │ Connectors  │  │ Connectors  │  │ Connectors  │ │   │
  │  │  │             │  │             │  │             │  │             │ │   │
  │  │  │ - Docker    │  │ - GitLab    │  │ - Slack     │  │ - Vault     │ │   │
  │  │  │ - ECR       │  │ - GitHub    │  │ - Teams     │  │ - AWS SM    │ │   │
  │  │  │ - ACR       │  │ - Jenkins   │  │ - Email     │  │ - Azure KV  │ │   │
  │  │  │ - Harbor    │  │ - Azure DO  │  │ - PagerDuty │  │             │ │   │
  │  │  └─────────────┘  └─────────────┘  └─────────────┘  └─────────────┘ │   │
  │  │                                                                      │   │
  │  └─────────────────────────────────────────────────────────────────────┘   │
  │                                                                             │
  └─────────────────────────────────────────────────────────────────────────────┘

Base Connector Interface

interface IConnector {
  // Metadata
  readonly typeId: string;
  readonly displayName: string;
  readonly version: string;
  readonly capabilities: ConnectorCapabilities;

  // Lifecycle
  initialize(config: IntegrationConfig): Promise<void>;
  dispose(): Promise<void>;

  // Health
  ping(config: IntegrationConfig): Promise<void>;
  healthCheck(config: IntegrationConfig, creds: Credential): Promise<HealthCheckResult>;

  // Authentication
  authenticate(config: IntegrationConfig, creds: Credential): Promise<AuthContext>;

  // Discovery (optional)
  discover?(
    config: IntegrationConfig,
    authContext: AuthContext,
    resourceType: string,
    filter?: DiscoveryFilter
  ): Promise<DiscoveredResource[]>;
}

interface ConnectorCapabilities {
  discovery: boolean;
  webhooks: boolean;
  streaming: boolean;
  batchOperations: boolean;
  customActions: string[];
}

Registry Connectors

IRegistryConnector

interface IRegistryConnector extends IConnector {
  // Repository operations
  listRepositories(authContext: AuthContext): Promise<Repository[]>;

  // Tag operations
  listTags(authContext: AuthContext, repository: string): Promise<Tag[]>;
  getManifest(authContext: AuthContext, repository: string, reference: string): Promise<Manifest>;
  getDigest(authContext: AuthContext, repository: string, tag: string): Promise<string>;

  // Image operations
  imageExists(authContext: AuthContext, repository: string, digest: string): Promise<boolean>;
  getImageMetadata(authContext: AuthContext, repository: string, digest: string): Promise<ImageMetadata>;
}

interface Repository {
  name: string;
  fullName: string;
  tagCount?: number;
  lastUpdated?: DateTime;
}

interface Tag {
  name: string;
  digest: string;
  createdAt?: DateTime;
  size?: number;
}

interface ImageMetadata {
  digest: string;
  mediaType: string;
  size: number;
  architecture: string;
  os: string;
  created: DateTime;
  labels: Record<string, string>;
  layers: LayerInfo[];
}

Docker Registry Connector

class DockerRegistryConnector implements IRegistryConnector {
  readonly typeId = "docker-registry";
  readonly displayName = "Docker Registry";
  readonly version = "1.0.0";
  readonly capabilities: ConnectorCapabilities = {
    discovery: true,
    webhooks: true,
    streaming: false,
    batchOperations: false,
    customActions: []
  };

  private httpClient: HttpClient;

  async initialize(config: DockerRegistryConfig): Promise<void> {
    this.httpClient = new HttpClient({
      baseUrl: config.url,
      timeout: config.timeout || 30000,
      insecureSkipVerify: config.insecureSkipVerify
    });
  }

  async ping(config: DockerRegistryConfig): Promise<void> {
    const response = await this.httpClient.get("/v2/");
    if (response.status !== 200 && response.status !== 401) {
      throw new Error(`Registry unavailable: ${response.status}`);
    }
  }

  async authenticate(
    config: DockerRegistryConfig,
    creds: BasicCredential
  ): Promise<AuthContext> {
    // Get auth challenge from /v2/
    const challenge = await this.getAuthChallenge();

    if (challenge.type === "bearer") {
      // OAuth2 token flow
      const token = await this.getToken(challenge, creds);
      return { type: "bearer", token };
    } else {
      // Basic auth
      return {
        type: "basic",
        credentials: Buffer.from(`${creds.username}:${creds.password}`).toString("base64")
      };
    }
  }

  async getDigest(
    authContext: AuthContext,
    repository: string,
    tag: string
  ): Promise<string> {
    const response = await this.httpClient.head(
      `/v2/${repository}/manifests/${tag}`,
      {
        headers: {
          ...this.authHeader(authContext),
          Accept: "application/vnd.docker.distribution.manifest.v2+json"
        }
      }
    );

    const digest = response.headers.get("docker-content-digest");
    if (!digest) {
      throw new Error("No digest header in response");
    }

    return digest;
  }

  async getImageMetadata(
    authContext: AuthContext,
    repository: string,
    digest: string
  ): Promise<ImageMetadata> {
    // Fetch manifest
    const manifest = await this.getManifest(authContext, repository, digest);

    // Fetch config blob
    const configDigest = manifest.config.digest;
    const configResponse = await this.httpClient.get(
      `/v2/${repository}/blobs/${configDigest}`,
      { headers: this.authHeader(authContext) }
    );

    const config = await configResponse.json();

    return {
      digest,
      mediaType: manifest.mediaType,
      size: manifest.config.size,
      architecture: config.architecture,
      os: config.os,
      created: new Date(config.created),
      labels: config.config?.Labels || {},
      layers: manifest.layers.map(l => ({
        digest: l.digest,
        size: l.size,
        mediaType: l.mediaType
      }))
    };
  }
}

ECR Connector

class ECRConnector implements IRegistryConnector {
  readonly typeId = "ecr";
  readonly displayName = "AWS ECR";
  readonly version = "1.0.0";
  readonly capabilities: ConnectorCapabilities = {
    discovery: true,
    webhooks: false,
    streaming: false,
    batchOperations: true,
    customActions: ["createRepository", "setLifecyclePolicy"]
  };

  private ecrClient: ECRClient;

  async initialize(config: ECRConfig): Promise<void> {
    this.ecrClient = new ECRClient({
      region: config.region,
      credentials: {
        accessKeyId: config.accessKeyId,
        secretAccessKey: config.secretAccessKey
      }
    });
  }

  async authenticate(
    config: ECRConfig,
    creds: AWSCredential
  ): Promise<AuthContext> {
    const command = new GetAuthorizationTokenCommand({});
    const response = await this.ecrClient.send(command);

    const authData = response.authorizationData?.[0];
    if (!authData?.authorizationToken) {
      throw new Error("Failed to get ECR authorization token");
    }

    return {
      type: "bearer",
      token: authData.authorizationToken,
      expiresAt: authData.expiresAt
    };
  }

  async listRepositories(authContext: AuthContext): Promise<Repository[]> {
    const repositories: Repository[] = [];
    let nextToken: string | undefined;

    do {
      const command = new DescribeRepositoriesCommand({
        nextToken
      });
      const response = await this.ecrClient.send(command);

      for (const repo of response.repositories || []) {
        repositories.push({
          name: repo.repositoryName!,
          fullName: repo.repositoryUri!,
          lastUpdated: repo.createdAt
        });
      }

      nextToken = response.nextToken;
    } while (nextToken);

    return repositories;
  }
}

CI/CD Connectors

ICICDConnector

interface ICICDConnector extends IConnector {
  // Pipeline operations
  listPipelines(authContext: AuthContext): Promise<Pipeline[]>;
  getPipeline(authContext: AuthContext, pipelineId: string): Promise<Pipeline>;

  // Trigger operations
  triggerPipeline(
    authContext: AuthContext,
    pipelineId: string,
    params: TriggerParams
  ): Promise<PipelineRun>;

  // Run operations
  getPipelineRun(authContext: AuthContext, runId: string): Promise<PipelineRun>;
  cancelPipelineRun(authContext: AuthContext, runId: string): Promise<void>;
  getPipelineRunLogs(authContext: AuthContext, runId: string): Promise<string>;
}

interface Pipeline {
  id: string;
  name: string;
  ref?: string;
  webUrl?: string;
}

interface TriggerParams {
  ref?: string;              // Branch/tag
  variables?: Record<string, string>;
}

interface PipelineRun {
  id: string;
  pipelineId: string;
  status: PipelineStatus;
  ref?: string;
  webUrl?: string;
  startedAt?: DateTime;
  finishedAt?: DateTime;
}

type PipelineStatus =
  | "pending"
  | "running"
  | "success"
  | "failed"
  | "cancelled";

GitLab CI Connector

class GitLabCIConnector implements ICICDConnector {
  readonly typeId = "gitlab-ci";
  readonly displayName = "GitLab CI/CD";
  readonly version = "1.0.0";
  readonly capabilities: ConnectorCapabilities = {
    discovery: true,
    webhooks: true,
    streaming: false,
    batchOperations: false,
    customActions: ["retryPipeline"]
  };

  private apiClient: GitLabClient;

  async initialize(config: GitLabCIConfig): Promise<void> {
    this.apiClient = new GitLabClient({
      baseUrl: config.url,
      projectId: config.projectId
    });
  }

  async authenticate(
    config: GitLabCIConfig,
    creds: TokenCredential
  ): Promise<AuthContext> {
    // Validate token with user endpoint
    this.apiClient.setToken(creds.token);
    await this.apiClient.get("/user");

    return {
      type: "bearer",
      token: creds.token
    };
  }

  async triggerPipeline(
    authContext: AuthContext,
    pipelineId: string,
    params: TriggerParams
  ): Promise<PipelineRun> {
    const response = await this.apiClient.post(
      `/projects/${this.projectId}/pipeline`,
      {
        ref: params.ref || this.defaultBranch,
        variables: Object.entries(params.variables || {}).map(([key, value]) => ({
          key,
          value,
          variable_type: "env_var"
        }))
      },
      { headers: { Authorization: `Bearer ${authContext.token}` } }
    );

    return {
      id: response.id.toString(),
      pipelineId: pipelineId,
      status: this.mapStatus(response.status),
      ref: response.ref,
      webUrl: response.web_url,
      startedAt: response.started_at ? new Date(response.started_at) : undefined
    };
  }

  async getPipelineRun(
    authContext: AuthContext,
    runId: string
  ): Promise<PipelineRun> {
    const response = await this.apiClient.get(
      `/projects/${this.projectId}/pipelines/${runId}`,
      { headers: { Authorization: `Bearer ${authContext.token}` } }
    );

    return {
      id: response.id.toString(),
      pipelineId: response.id.toString(),
      status: this.mapStatus(response.status),
      ref: response.ref,
      webUrl: response.web_url,
      startedAt: response.started_at ? new Date(response.started_at) : undefined,
      finishedAt: response.finished_at ? new Date(response.finished_at) : undefined
    };
  }

  private mapStatus(gitlabStatus: string): PipelineStatus {
    const statusMap: Record<string, PipelineStatus> = {
      created: "pending",
      waiting_for_resource: "pending",
      preparing: "pending",
      pending: "pending",
      running: "running",
      success: "success",
      failed: "failed",
      canceled: "cancelled",
      skipped: "cancelled",
      manual: "pending"
    };
    return statusMap[gitlabStatus] || "pending";
  }
}

Notification Connectors

INotificationConnector

interface INotificationConnector extends IConnector {
  // Channel operations
  listChannels(authContext: AuthContext): Promise<Channel[]>;

  // Send operations
  sendMessage(
    authContext: AuthContext,
    channel: string,
    message: NotificationMessage
  ): Promise<MessageResult>;

  sendTemplate(
    authContext: AuthContext,
    channel: string,
    templateId: string,
    data: Record<string, any>
  ): Promise<MessageResult>;
}

interface Channel {
  id: string;
  name: string;
  type: string;
}

interface NotificationMessage {
  text: string;
  title?: string;
  color?: string;
  fields?: MessageField[];
  actions?: MessageAction[];
}

interface MessageField {
  name: string;
  value: string;
  inline?: boolean;
}

interface MessageAction {
  type: "button" | "link";
  text: string;
  url?: string;
  style?: "primary" | "danger" | "default";
}

Slack Connector

class SlackConnector implements INotificationConnector {
  readonly typeId = "slack";
  readonly displayName = "Slack";
  readonly version = "1.0.0";
  readonly capabilities: ConnectorCapabilities = {
    discovery: true,
    webhooks: true,
    streaming: false,
    batchOperations: false,
    customActions: ["addReaction", "updateMessage"]
  };

  private slackClient: WebClient;

  async initialize(config: SlackConfig): Promise<void> {
    // Client initialized in authenticate
  }

  async authenticate(
    config: SlackConfig,
    creds: TokenCredential
  ): Promise<AuthContext> {
    this.slackClient = new WebClient(creds.token);

    // Test authentication
    const result = await this.slackClient.auth.test();
    if (!result.ok) {
      throw new Error("Slack authentication failed");
    }

    return {
      type: "bearer",
      token: creds.token,
      teamId: result.team_id,
      userId: result.user_id
    };
  }

  async listChannels(authContext: AuthContext): Promise<Channel[]> {
    const channels: Channel[] = [];
    let cursor: string | undefined;

    do {
      const result = await this.slackClient.conversations.list({
        types: "public_channel,private_channel",
        cursor
      });

      for (const channel of result.channels || []) {
        channels.push({
          id: channel.id!,
          name: channel.name!,
          type: channel.is_private ? "private" : "public"
        });
      }

      cursor = result.response_metadata?.next_cursor;
    } while (cursor);

    return channels;
  }

  async sendMessage(
    authContext: AuthContext,
    channel: string,
    message: NotificationMessage
  ): Promise<MessageResult> {
    const blocks = this.buildBlocks(message);

    const result = await this.slackClient.chat.postMessage({
      channel,
      text: message.text,
      blocks,
      attachments: message.color ? [{
        color: message.color,
        blocks
      }] : undefined
    });

    return {
      messageId: result.ts!,
      channel: result.channel!,
      success: result.ok
    };
  }

  private buildBlocks(message: NotificationMessage): KnownBlock[] {
    const blocks: KnownBlock[] = [];

    if (message.title) {
      blocks.push({
        type: "header",
        text: {
          type: "plain_text",
          text: message.title
        }
      });
    }

    blocks.push({
      type: "section",
      text: {
        type: "mrkdwn",
        text: message.text
      }
    });

    if (message.fields?.length) {
      blocks.push({
        type: "section",
        fields: message.fields.map(f => ({
          type: "mrkdwn",
          text: `*${f.name}*\n${f.value}`
        }))
      });
    }

    if (message.actions?.length) {
      blocks.push({
        type: "actions",
        elements: message.actions.map(a => ({
          type: "button",
          text: {
            type: "plain_text",
            text: a.text
          },
          url: a.url,
          style: a.style === "danger" ? "danger" : "primary"
        }))
      });
    }

    return blocks;
  }
}

Secret Store Connectors

ISecretConnector

interface ISecretConnector extends IConnector {
  // Secret operations
  getSecret(
    authContext: AuthContext,
    path: string,
    key?: string
  ): Promise<SecretValue>;

  listSecrets(
    authContext: AuthContext,
    path: string
  ): Promise<string[]>;
}

interface SecretValue {
  value: string;
  version?: string;
  createdAt?: DateTime;
  expiresAt?: DateTime;
}

HashiCorp Vault Connector

class VaultConnector implements ISecretConnector {
  readonly typeId = "hashicorp-vault";
  readonly displayName = "HashiCorp Vault";
  readonly version = "1.0.0";
  readonly capabilities: ConnectorCapabilities = {
    discovery: true,
    webhooks: false,
    streaming: false,
    batchOperations: false,
    customActions: ["renewToken"]
  };

  private vaultClient: VaultClient;

  async initialize(config: VaultConfig): Promise<void> {
    this.vaultClient = new VaultClient({
      endpoint: config.url,
      namespace: config.namespace
    });
  }

  async authenticate(
    config: VaultConfig,
    creds: Credential
  ): Promise<AuthContext> {
    let token: string;

    switch (config.authMethod) {
      case "token":
        token = (creds as TokenCredential).token;
        break;

      case "approle":
        const approle = creds as AppRoleCredential;
        const result = await this.vaultClient.auth.approle.login({
          role_id: approle.roleId,
          secret_id: approle.secretId
        });
        token = result.auth.client_token;
        break;

      case "kubernetes":
        throw new Error("Kubernetes auth is unsupported for Stella Ops deployment connectors");

      default:
        throw new Error(`Unsupported auth method: ${config.authMethod}`);
    }

    this.vaultClient.token = token;

    return {
      type: "bearer",
      token,
      renewable: true
    };
  }

  async getSecret(
    authContext: AuthContext,
    path: string,
    key?: string
  ): Promise<SecretValue> {
    const result = await this.vaultClient.kv.v2.read({
      mount_path: this.mountPath,
      path
    });

    const data = result.data.data;
    const value = key ? data[key] : JSON.stringify(data);

    return {
      value,
      version: result.data.metadata.version.toString(),
      createdAt: new Date(result.data.metadata.created_time)
    };
  }

  async listSecrets(
    authContext: AuthContext,
    path: string
  ): Promise<string[]> {
    const result = await this.vaultClient.kv.v2.list({
      mount_path: this.mountPath,
      path
    });

    return result.data.keys;
  }
}

Custom Connector Development

Plugin Structure

my-connector/
  ├── manifest.yaml
  ├── src/
  │   ├── connector.ts
  │   ├── config.ts
  │   └── types.ts
  └── package.json

Manifest

# manifest.yaml
id: my-custom-connector
version: 1.0.0
name: My Custom Connector
description: Custom connector for XYZ service
author: Your Name

connector:
  typeId: my-service
  displayName: My Service
  entrypoint: ./src/connector.js

  capabilities:
    discovery: true
    webhooks: false
    streaming: false
    batchOperations: false

  config_schema:
    type: object
    properties:
      url:
        type: string
        format: uri
        description: Service URL
      timeout:
        type: integer
        default: 30000
    required:
      - url

  credential_types:
    - api-key
    - oauth2

Implementation

// connector.ts
import { IConnector, ConnectorCapabilities } from "@stella-ops/connector-sdk";

export class MyConnector implements IConnector {
  readonly typeId = "my-service";
  readonly displayName = "My Service";
  readonly version = "1.0.0";
  readonly capabilities: ConnectorCapabilities = {
    discovery: true,
    webhooks: false,
    streaming: false,
    batchOperations: false,
    customActions: []
  };

  async initialize(config: MyConfig): Promise<void> {
    // Initialize your connector
  }

  async dispose(): Promise<void> {
    // Cleanup resources
  }

  async ping(config: MyConfig): Promise<void> {
    // Check connectivity
  }

  async healthCheck(config: MyConfig, creds: Credential): Promise<HealthCheckResult> {
    // Full health check
  }

  async authenticate(config: MyConfig, creds: Credential): Promise<AuthContext> {
    // Authenticate and return context
  }

  async discover(
    config: MyConfig,
    authContext: AuthContext,
    resourceType: string,
    filter?: DiscoveryFilter
  ): Promise<DiscoveredResource[]> {
    // Discover resources
  }
}

// Export connector factory
export default function createConnector(): IConnector {
  return new MyConnector();
}

References

• Integrations Overview
• Webhooks
• Plugin System