coreflow/ts/coreflow.classes.clustermanager.ts

import * as plugins from './coreflow.plugins.js';
import { logger } from './coreflow.logging.js';
import { Coreflow } from './coreflow.classes.coreflow.js';

export class ClusterManager {
  public coreflowRef: Coreflow;
  public configSubscription?: plugins.smartrx.rxjs.Subscription;

  public readyDeferred = plugins.smartpromise.defer();

  public commonDockerData = {
    networkNames: {
      sznWebgateway: 'sznwebgateway',
      sznCorechat: 'szncorechat',
    },
  };

  constructor(coreflowRefArg: Coreflow) {
    this.coreflowRef = coreflowRefArg;
  }

  private async getDockerServiceByName(serviceNameArg: string): Promise<plugins.docker.DockerService | null> {
    try {
      return await this.coreflowRef.dockerHost.getServiceByName(serviceNameArg);
    } catch (error) {
      if ((error as Error).message === `Service not found: ${serviceNameArg}`) {
        return null;
      }
      throw error;
    }
  }

  private getDockerSafeName(valueArg: string, maxLengthArg = 64) {
    const safeName = valueArg
      .replace(/[^a-zA-Z0-9_.-]+/g, '-')
      .replace(/^[^a-zA-Z0-9]+|[^a-zA-Z0-9]+$/g, '')
      .slice(0, maxLengthArg)
      .replace(/[^a-zA-Z0-9]+$/g, '');
    return safeName || 'resource';
  }

  private getWorkloadSecretName(serviceArgFromCloudly: plugins.servezoneInterfaces.data.IService) {
    const serviceName = this.getDockerSafeName(serviceArgFromCloudly.data.name, 36);
    const serviceId = this.getDockerSafeName(serviceArgFromCloudly.id, 20);
    return this.getDockerSafeName(`${serviceName}-${serviceId}-secret`);
  }

  private getWorkloadServiceDeploymentLabels(
    serviceArgFromCloudly: plugins.servezoneInterfaces.data.IService,
    containerImageFromCloudly: plugins.servezoneInterfaces.data.IImage,
  ) {
    const desiredImageVersion =
      serviceArgFromCloudly.data.imageVersion ||
      serviceArgFromCloudly.data.registryTarget?.tag ||
      'latest';
    const desiredImageVersionData = (containerImageFromCloudly.data.versions || []).find((versionArg) => {
      return versionArg.versionString === desiredImageVersion;
    });
    const desiredRegistryDigest = desiredImageVersionData?.digest || (
      containerImageFromCloudly.data.lastPushEvent?.tag === desiredImageVersion
        ? containerImageFromCloudly.data.lastPushEvent.digest
        : ''
    );

    return {
      'serve.zone.serviceId': serviceArgFromCloudly.id,
      'serve.zone.imageId': serviceArgFromCloudly.data.imageId || '',
      'serve.zone.imageVersion': desiredImageVersion,
      'serve.zone.registryImageUrl': serviceArgFromCloudly.data.registryTarget?.imageUrl || '',
      'serve.zone.registryDigest': desiredRegistryDigest || '',
    };
  }

  private async pullRegistryTargetImage(
    registryTargetArg: plugins.servezoneInterfaces.data.IRegistryTarget,
  ): Promise<plugins.docker.DockerImage> {
    const registryImageName = `${registryTargetArg.registryHost}/${registryTargetArg.repository}`;
    const registryImageTag = registryTargetArg.tag || 'latest';
    const registryImageRef = `${registryImageName}:${registryImageTag}`;
    const response = await this.coreflowRef.dockerHost.request(
      'POST',
      `/images/create?fromImage=${encodeURIComponent(registryImageName)}&tag=${encodeURIComponent(
        registryImageTag,
      )}`,
    );
    if (response.statusCode >= 300) {
      const existingImage = await this.coreflowRef.dockerHost.getImageByName(registryImageRef);
      if (existingImage) {
        logger.log(
          'warn',
          `registry pull failed for ${registryImageRef}, using locally cached image`,
        );
        return existingImage;
      }
      throw new Error(`Failed to pull registry image ${registryImageRef}`);
    }

    const localDockerImage = await this.coreflowRef.dockerHost.getImageByName(registryImageRef);
    if (!localDockerImage) {
      throw new Error(`Registry image ${registryImageRef} not found after pull`);
    }
    return localDockerImage;
  }

  /**
   * starts the cluster manager
   */
  public async start() {
    const config = await this.coreflowRef.cloudlyConnector.getConfigFromCloudly();
    this.readyDeferred.resolve();

    // subscriptions
    // this subscription is the start point for most updates on the cluster
    this.configSubscription =
      this.coreflowRef.cloudlyConnector.cloudlyApiClient.configUpdateSubject.subscribe(
        async (dataArg) => {
          this.coreflowRef.taskManager.updateBaseServicesTask.trigger();
        },
      );
  }

  /**
   * stops the clustermanager
   */
  public async stop() {
    this.configSubscription ? this.configSubscription.unsubscribe() : null;
  }

  /**
   * provisions base services
   */
  public async provisionBaseServices() {
    // swarm should be enabled by lower level serverconfig package
    // get current situation
    const networks = await this.coreflowRef.dockerHost.listNetworks();
    logger.log('info', 'There are currently ' + networks.length + ' networks');
    for (const network of networks) {
      logger.log('info', 'Network: ' + network.Name);
    }

    // make sure there is a network for the webgateway
    let sznWebgatewayNetwork = await this.coreflowRef.dockerHost.getNetworkByName(
      this.commonDockerData.networkNames.sznWebgateway,
    );

    if (!sznWebgatewayNetwork) {
      logger.log('info', 'Creating network: ' + this.commonDockerData.networkNames.sznWebgateway);
      sznWebgatewayNetwork = await this.coreflowRef.dockerHost.createNetwork({
        Name: this.commonDockerData.networkNames.sznWebgateway,
      });
    } else {
      logger.log('ok', 'sznWebgateway is already present');
    }

    // corechat network so base services can talk to each other
    let sznCorechatNetwork = await this.coreflowRef.dockerHost.getNetworkByName(
      this.commonDockerData.networkNames.sznCorechat,
    );
    if (!sznCorechatNetwork) {
      sznCorechatNetwork = await this.coreflowRef.dockerHost.createNetwork({
        Name: this.commonDockerData.networkNames.sznCorechat,
      });
    } else {
      logger.log('ok', 'sznCorechat is already present');
    }

    logger.log('info', `Waiting for networks to be ready...`);
    await plugins.smartdelay.delayFor(10000);

    logger.log('success', `Successfully created networks for servezone`);

    // Images
    logger.log('info', `now updating docker images of base services...`);

    const coretrafficImage = await this.coreflowRef.dockerHost.createImageFromRegistry({
      imageUrl: 'code.foss.global/serve.zone/coretraffic',
    });

    const corelogImage = await this.coreflowRef.dockerHost.createImageFromRegistry({
      imageUrl: 'code.foss.global/serve.zone/corelog',
    });

    // SERVICES
    // lets deploy the base services
    // coretraffic
    let coretrafficService: plugins.docker.DockerService | null;
    coretrafficService = await this.getDockerServiceByName(
      'coretraffic',
    );

    if (coretrafficService && (await coretrafficService.needsUpdate())) {
      await coretrafficService.remove();
      await plugins.smartdelay.delayFor(5000);
      coretrafficService = null;
    } else {
      logger.log('ok', `coretraffic service is up to date`);
    }

    if (!coretrafficService) {
      coretrafficService = await this.coreflowRef.dockerHost.createService({
        image: coretrafficImage,
        labels: {},
        name: 'coretraffic',
        networks: [sznCorechatNetwork, sznWebgatewayNetwork],
        networkAlias: 'coretraffic',
        ports: ['80:7999', '443:8000'],
        secrets: [],
        resources: {
          memorySizeMB: 1100,
          volumeMounts: [],
        },
      });
    } else {
      logger.log('ok', 'coretraffic service is already present');
    }
    logger.log('info', 'wait for coretraffic to be up and running');
    await plugins.smartdelay.delayFor(10000);

    // corelog
    let corelogService: plugins.docker.DockerService | null;
    corelogService = await this.getDockerServiceByName(
      'corelog',
    );

    if (corelogService && (await corelogService.needsUpdate())) {
      await corelogService.remove();
      corelogService = null;
    } else {
      logger.log('ok', `corelog service is up to date`);
    }

    if (!corelogService) {
      corelogService = await this.coreflowRef.dockerHost.createService({
        image: corelogImage,
        labels: {},
        name: 'corelog',
        networks: [sznCorechatNetwork],
        networkAlias: 'corelog',
        ports: [],
        secrets: [],
        resources: {
          memorySizeMB: 120,
          volumeMounts: [],
        },
      });
    } else {
      logger.log('ok', 'corelog service is already present');
    }
    logger.log('info', 'waiting for corelog to be up and running');
    await plugins.smartdelay.delayFor(10000);
  }

  public async provisionWorkloadService(
    serviceArgFromCloudly: plugins.servezoneInterfaces.data.IService,
  ) {
    logger.log(
      'info',
      `deploying service ${serviceArgFromCloudly.data.name}@${serviceArgFromCloudly.data.imageVersion}...`,
    );

    // get the image from cloudly
    logger.log(
      'info',
      `getting image for ${serviceArgFromCloudly.data.name}@${serviceArgFromCloudly.data.imageVersion}`,
    );
    const containerImageFromCloudly =
      await this.coreflowRef.cloudlyConnector.cloudlyApiClient.image.getImageById(
        serviceArgFromCloudly.data.imageId,
      );
    const deploymentLabels = this.getWorkloadServiceDeploymentLabels(
      serviceArgFromCloudly,
      containerImageFromCloudly,
    );
    let localDockerImage: plugins.docker.DockerImage;

    // lets get the docker image for the service
    if (serviceArgFromCloudly.data.registryTarget) {
      await this.coreflowRef.dockerHost.auth({
        username: this.coreflowRef.cloudlyConnector.identity.name,
        password: this.coreflowRef.cloudlyConnector.coreflowJumpCode,
        serveraddress: serviceArgFromCloudly.data.registryTarget.registryHost,
      });
      localDockerImage = await this.pullRegistryTargetImage(serviceArgFromCloudly.data.registryTarget);
    } else if (containerImageFromCloudly.data.location?.internal) {
      const imageStream = await containerImageFromCloudly.pullImageVersion(
        serviceArgFromCloudly.data.imageVersion,
      );
      localDockerImage = await this.coreflowRef.dockerHost.createImageFromTarStream(
        plugins.smartstream.nodewebhelpers.convertWebReadableToNodeReadable(imageStream),
        {
          imageUrl: containerImageFromCloudly.id,
          imageTag: serviceArgFromCloudly.data.imageVersion,
        },
      );
    } else if (
      containerImageFromCloudly.data.location?.externalRegistryId &&
      containerImageFromCloudly.data.location?.externalImageTag
    ) {
      const externalRegistry =
        await this.coreflowRef.cloudlyConnector.cloudlyApiClient.externalRegistry.getRegistryById(
          containerImageFromCloudly.data.location.externalRegistryId,
        );
      // Lets authenticate against the external registry
      // TODO: deduplicate this, check wether we are already authenticated
      if (!externalRegistry.data.username || !externalRegistry.data.password) {
        throw new Error(`External registry ${externalRegistry.id} is missing credentials`);
      }
      await this.coreflowRef.dockerHost.auth({
        username: externalRegistry.data.username,
        password: externalRegistry.data.password,
        serveraddress: externalRegistry.data.url,
      });
      localDockerImage = await this.coreflowRef.dockerHost.createImageFromRegistry({
        imageUrl: containerImageFromCloudly.id,
        imageTag: serviceArgFromCloudly.data.imageVersion,
      });
      await localDockerImage.pullLatestImageFromRegistry();
    } else {
      throw new Error('Invalid image location');
    }

    let containerService: plugins.docker.DockerService | null = await this.getDockerServiceByName(
      serviceArgFromCloudly.data.name,
    );
    this.coreflowRef.cloudlyConnector.cloudlyApiClient;

    const dockerSecretName = this.getWorkloadSecretName(serviceArgFromCloudly);
    let containerSecret: plugins.docker.DockerSecret | undefined | null = await this.coreflowRef.dockerHost.getSecretByName(
      dockerSecretName,
    );

    // existing network to connect to
    const webGatewayNetwork = await this.coreflowRef.dockerHost.getNetworkByName(
      this.commonDockerData.networkNames.sznWebgateway,
    );
    if (!webGatewayNetwork) {
      throw new Error(`Missing required Docker network ${this.commonDockerData.networkNames.sznWebgateway}`);
    }

    if (containerService) {
      const existingLabels = containerService.Spec.Labels || {};
      const cloudlyDeploymentLabelsChanged = Object.entries(deploymentLabels).some(([key, value]) => {
        return existingLabels[key] !== value;
      });
      const dockerImageNeedsUpdate = serviceArgFromCloudly.data.registryTarget
        ? false
        : await containerService.needsUpdate();

      if (cloudlyDeploymentLabelsChanged || dockerImageNeedsUpdate) {
        logger.log('info', `service ${serviceArgFromCloudly.data.name} desired state changed, recreating`);
        await containerService.remove();
        if (containerSecret) {
          await containerSecret.remove();
        }
        containerService = null;
        containerSecret = null;
      }
    }

    if (!containerService) {
      containerSecret = await this.coreflowRef.dockerHost.getSecretByName(
        dockerSecretName,
      );
      if (containerSecret) {
        await containerSecret.remove();
      }

      const secretBundle =
        await this.coreflowRef.cloudlyConnector.cloudlyApiClient.secretbundle.getSecretBundleById(
          serviceArgFromCloudly.data.secretBundleId,
        );

      // lets create the relevant stuff on the docker side
      containerSecret = await this.coreflowRef.dockerHost.createSecret({
        name: dockerSecretName,
        contentArg: JSON.stringify(await secretBundle.getFlatKeyValueObjectForEnvironment()),
        labels: {},
        version: serviceArgFromCloudly.data.imageVersion,
      });
      containerService = await this.coreflowRef.dockerHost.createService({
        name: serviceArgFromCloudly.data.name,
        image: localDockerImage,
        networks: [webGatewayNetwork],
        secrets: [containerSecret],
        ports: [],
        labels: deploymentLabels,
        resources: serviceArgFromCloudly.data.resources,
        // TODO: introduce a clean name here, that is guaranteed to work with APIs.
        networkAlias: serviceArgFromCloudly.data.name,
      });
    }
  }

  public async provisionWorkloadServices(
    _clusterConfigArg: plugins.servezoneInterfaces.data.ICluster,
  ) {
    const services =
      (await this.coreflowRef.cloudlyConnector.cloudlyApiClient.services.getServices()) as unknown as plugins.servezoneInterfaces.data.IService[];
    for (const service of services) {
      if (service.data.serviceCategory && service.data.serviceCategory !== 'workload') {
        continue;
      }
      await this.provisionWorkloadService(service);
    }
  }

  /**
   * update traffic routing
   */
  public async updateTrafficRouting(
    _clusterConfigArg: plugins.servezoneInterfaces.data.ICluster,
  ) {
    const services = await this.coreflowRef.dockerHost.listServices();
    const webGatewayNetwork = await this.coreflowRef.dockerHost.getNetworkByName(
      this.commonDockerData.networkNames.sznWebgateway,
    );
    if (!webGatewayNetwork) {
      throw new Error(`Missing required Docker network ${this.commonDockerData.networkNames.sznWebgateway}`);
    }
    const reverseProxyConfigs: plugins.servezoneInterfaces.data.IReverseProxyConfig[] = [];

    const pushProxyConfig = async (
      serviceNameArg: string,
      hostNameArg: string,
      containerDestinationIp: string,
      webDestinationPort: string,
    ) => {
      logger.log('ok', `trying to obtain a certificate for ${hostNameArg}`);
      const certificate =
        await this.coreflowRef.cloudlyConnector.getCertificateForDomainFromCloudly(hostNameArg);
      reverseProxyConfigs.push({
        destinationIps: [containerDestinationIp],
        destinationPorts: [Number(webDestinationPort)],
        hostName: hostNameArg,
        privateKey: certificate.privateKey,
        publicKey: certificate.publicKey,
      });
      logger.log(
        'success',
        `pushed routing config for ${hostNameArg} on workload service ${serviceNameArg}`,
      );
    };

    logger.log('info', `Found ${services.length} services!`);
    const workloadServices =
      (await this.coreflowRef.cloudlyConnector.cloudlyApiClient.services.getServices()) as unknown as plugins.servezoneInterfaces.data.IService[];
    for (const service of services) {
      let workloadConfig: plugins.servezoneInterfaces.data.IService | undefined;
      for (const workloadServiceConfig of workloadServices) {
        if (service.Spec.Name === workloadServiceConfig.data.name) {
          workloadConfig = workloadServiceConfig;
          break;
        }
      }

      if (workloadConfig) {
        logger.log('ok', `found workload service ${service.Spec.Name}`);
        const containersOfServicesOnNetwork = (
          await webGatewayNetwork.getContainersOnNetworkForService(service)
        ).filter((container) => container.Name !== service.Spec.Name);
        // TODO: make this multi container ready
        if (!containersOfServicesOnNetwork[0]) {
          logger.log(
            'error',
            `There seems to be no container available for service ${service.Spec.Name}`,
          );
          continue;
        }
        const containerDestinationIp = containersOfServicesOnNetwork[0].IPv4Address.split('/')[0];
        const hostNames = (workloadConfig.data.domains || []).map((domainEntry) => domainEntry.name);

        // lets route the web port
        const webDestinationPort: string = workloadConfig.data.ports.web.toString();
        for (const hostName of hostNames) {
          await pushProxyConfig(
            workloadConfig.data.name,
            hostName,
            containerDestinationIp,
            webDestinationPort,
          );
        }

        // lets route custom ports
        if (workloadConfig.data.ports.custom) {
          const customDomainKeys = Object.keys(workloadConfig.data.ports.custom);
          for (const customDomainKey of customDomainKeys) {
            await pushProxyConfig(
              workloadConfig.data.name,
              customDomainKey,
              containerDestinationIp,
              workloadConfig.data.ports.custom[customDomainKey],
            );
          }
        }
      } else {
        logger.log(
          'ok',
          `service ${service.Spec.Name} is not a workload service and won't receive traffic`,
        );
      }
    }
    console.log(reverseProxyConfigs);
    await this.coreflowRef.corechatConnector.setReverseConfigs(reverseProxyConfigs);
  }
}