import * as plugins from './coreflow.plugins.js'; import { logger } from './coreflow.logging.js'; import { Coreflow } from './coreflow.classes.coreflow.js'; import type { IExternalGatewayConfig } from './coreflow.connector.externalgateway.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 { 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 { 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.requests.config.IRequest_Any_Cloudly_GetClusterConfig['response'] & { externalGateway?: IExternalGatewayConfig; }, ) { const externalGatewayConfig = clusterConfigArg.externalGateway; 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 ( workloadServiceArg: plugins.servezoneInterfaces.data.IService, hostNameArg: string, containerDestinationIp: string, webDestinationPort: string, ) => { logger.log('ok', `trying to obtain a certificate for ${hostNameArg}`); let certificate = await this.coreflowRef.externalGatewayConnector.exportCertificateForDomain( externalGatewayConfig, hostNameArg, ).catch((error) => { logger.log('warn', `external gateway certificate export failed for ${hostNameArg}: ${(error as Error).message}`); return undefined; }); certificate = 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 ${workloadServiceArg.data.name}`, ); await this.coreflowRef.externalGatewayConnector.syncWorkAppRoute({ config: externalGatewayConfig, service: workloadServiceArg, hostname: hostNameArg, }).catch((error) => { logger.log('warn', `external gateway route sync failed for ${hostNameArg}: ${(error as Error).message}`); }); }; 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, 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, 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); } }