more

src/resources/cert-manager/certificate/certificate.ts

@@ -3,6 +3,7 @@ import type { K8SCertificateV1 } from 'src/__generated__/resources/K8SCertificat
 
 import { CRD } from '#resources/core/crd/crd.ts';
 import { Resource, ResourceService, type ResourceOptions } from '#services/resources/resources.ts';
+import { NotReadyError } from '#utils/errors.ts';
 
 class Certificate extends Resource<KubernetesObject & K8SCertificateV1> {
   public static readonly apiVersion = 'cert-manager.io/v1';
@@ -18,12 +19,19 @@ class Certificate extends Resource<KubernetesObject & K8SCertificateV1> {
   }
 
   #handleCrdChanged = () => {
-    this.emit('changed');
+    this.emit('changed', this.manifest);
   };
 
   public get hasCRD() {
     return this.#crd.exists;
   }
+
+  public set = async (manifest: KubernetesObject & K8SCertificateV1) => {
+    if (!this.hasCRD) {
+      throw new NotReadyError('MissingCRD', 'certificates.cert-manager.io does not exist');
+    }
+    return this.ensure(manifest);
+  };
 }
 
 export { Certificate };

src/resources/core/pvc/pvc.ts

@@ -3,7 +3,7 @@ import type { V1PersistentVolumeClaim } from '@kubernetes/client-node';
 import { StorageClass } from '../storage-class/storage-class.ts';
 import { PersistentVolume } from '../pv/pv.ts';
 
-import { Resource, ResourceService } from '#services/resources/resources.ts';
+import { Resource, ResourceService, type ResourceOptions } from '#services/resources/resources.ts';
 
 const PROVISIONER = 'homelab-operator';
 
@@ -11,8 +11,14 @@ class PVC extends Resource<V1PersistentVolumeClaim> {
   public static readonly apiVersion = 'v1';
   public static readonly kind = 'PersistentVolumeClaim';
 
+  constructor(options: ResourceOptions<V1PersistentVolumeClaim>) {
+    super(options);
+    this.on('changed', this.reconcile);
+  }
+
   public reconcile = async () => {
     const storageClassName = this.spec?.storageClassName;
+    console.log('PVC', this.name, storageClassName);
     if (!storageClassName) {
       return;
     }

src/resources/homelab/authentik-server/authentik-server.ts

@@ -16,6 +16,8 @@ import { Service } from '#resources/core/service/service.ts';
 import { HelmRelease } from '#resources/flux/helm-release/helm-release.ts';
 import { RepoService } from '#bootstrap/repos/repos.ts';
 import { VirtualService } from '#resources/istio/virtual-service/virtual-service.ts';
+import { DestinationRule } from '#resources/istio/destination-rule/destination-rule.ts';
+import { NotReadyError } from '#utils/errors.ts';
 
 const specSchema = z.object({
   environment: z.string(),
@@ -43,6 +45,7 @@ class AuthentikServer extends CustomResource<typeof specSchema> {
   #service: Service;
   #helmRelease: HelmRelease;
   #virtualService: VirtualService;
+  #destinationRule: DestinationRule;
 
   constructor(options: CustomResourceOptions<typeof specSchema>) {
     super(options);
@@ -67,17 +70,20 @@ class AuthentikServer extends CustomResource<typeof specSchema> {
 
     this.#virtualService = resourceService.get(VirtualService, this.name, this.namespace);
     this.#virtualService.on('changed', this.queueReconcile);
+
+    this.#destinationRule = resourceService.get(DestinationRule, this.name, this.namespace);
+    this.#destinationRule.on('changed', this.queueReconcile);
   }
 
   public reconcile = async () => {
     if (!this.spec) {
-      return;
+      throw new NotReadyError('MissingSpec');
     }
     const resourceService = this.services.get(ResourceService);
 
     this.#environment.current = resourceService.get(Environment, this.spec.environment);
     if (!this.#environment.current.spec) {
-      return;
+      throw new NotReadyError('MissingEnvSpev');
     }
 
     await this.#database.ensure({
@@ -91,7 +97,7 @@ class AuthentikServer extends CustomResource<typeof specSchema> {
 
     const databaseSecret = this.#database.secret.value;
     if (!databaseSecret) {
-      return;
+      throw new NotReadyError('MissingDatabaseSecret');
     }
 
     await this.#initSecret.set(
@@ -111,7 +117,7 @@ class AuthentikServer extends CustomResource<typeof specSchema> {
 
     const initSecret = this.#initSecret.value;
     if (!initSecret) {
-      return;
+      throw new NotReadyError('MissingInitSecret');
     }
 
     const domain = `${this.spec?.subdomain || 'authentik'}.${this.#environment.current.spec.domain}`;
@@ -129,7 +135,7 @@ class AuthentikServer extends CustomResource<typeof specSchema> {
     );
     const secret = this.#secret.value;
     if (!secret) {
-      return;
+      throw new NotReadyError('MissingSecret');
     }
 
     const repoService = this.services.get(RepoService);
@@ -214,6 +220,20 @@ class AuthentikServer extends CustomResource<typeof specSchema> {
       },
     });
 
+    await this.#destinationRule.ensure({
+      metadata: {
+        ownerReferences: [this.ref],
+      },
+      spec: {
+        host: this.#service.hostname,
+        trafficPolicy: {
+          tls: {
+            mode: 'DISABLE',
+          },
+        },
+      },
+    });
+
     const gateway = this.#environment.current.gateway;
     await this.#virtualService.set({
       metadata: {

src/resources/homelab/environment/environment.ts

@@ -11,6 +11,7 @@ import { Certificate } from '#resources/cert-manager/certificate/certificate.ts'
 import { StorageClass } from '#resources/core/storage-class/storage-class.ts';
 import { PROVISIONER } from '#resources/core/pvc/pvc.ts';
 import { Gateway } from '#resources/istio/gateway/gateway.ts';
+import { NotReadyError } from '#utils/errors.ts';
 
 const specSchema = z.object({
   domain: z.string(),
@@ -86,7 +87,7 @@ class Environment extends CustomResource<typeof specSchema> {
   public reconcile = async () => {
     const { data: spec, success } = specSchema.safeParse(this.spec);
     if (!success || !spec) {
-      return;
+      throw new NotReadyError('InvalidSpec');
     }
     await this.#namespace.ensure({
       metadata: {
@@ -95,24 +96,22 @@ class Environment extends CustomResource<typeof specSchema> {
         },
       },
     });
-    if (this.#certificate.hasCRD) {
-      await this.#certificate.ensure({
-        metadata: {
-          ownerReferences: [this.ref],
+    await this.#certificate.ensure({
+      metadata: {
+        ownerReferences: [this.ref],
+      },
+      spec: {
+        secretName: `${this.name}-tls`,
+        issuerRef: {
+          name: spec.tls.issuer,
+          kind: 'ClusterIssuer',
         },
-        spec: {
-          secretName: `${this.name}-tls`,
-          issuerRef: {
-            name: spec.tls.issuer,
-            kind: 'ClusterIssuer',
-          },
-          dnsNames: [`*.${spec.domain}`],
-          privateKey: {
-            rotationPolicy: 'Always',
-          },
+        dnsNames: [`*.${spec.domain}`],
+        privateKey: {
+          rotationPolicy: 'Always',
         },
-      });
-    }
+      },
+    });
     await this.#storageClass.ensure({
       metadata: {
         ownerReferences: [this.ref],

src/resources/homelab/postgres-database/postgres-database.ts

@@ -12,6 +12,8 @@ import { Secret } from '#resources/core/secret/secret.ts';
 import { API_VERSION } from '#utils/consts.ts';
 import { getWithNamespace } from '#utils/naming.ts';
 import { PostgresService } from '#services/postgres/postgres.service.ts';
+import { NotReadyError } from '#utils/errors.ts';
+import { generateRandomHexPass } from '#utils/secrets.ts';
 
 const specSchema = z.object({
   environment: z.string().optional(),
@@ -77,17 +79,17 @@ class PostgresDatabase extends CustomResource<typeof specSchema> {
       this.#cluster.current = environment.postgresCluster;
     } else {
       this.#cluster.current = undefined;
-      return;
+      throw new NotReadyError('MissingEnvOrClusterSpec');
     }
 
     const clusterSecret = this.#cluster.current.secret.value;
     if (!clusterSecret) {
-      return;
+      throw new NotReadyError('MissingClusterSecret');
     }
 
     await this.#secret.set(
       (current) => ({
-        password: crypto.randomUUID(),
+        password: generateRandomHexPass(),
         user: this.username,
         database: this.database,
         ...current,
@@ -103,7 +105,7 @@ class PostgresDatabase extends CustomResource<typeof specSchema> {
 
     const secret = this.#secret.value;
     if (!secret) {
-      return;
+      throw new NotReadyError('MissingSecret');
     }
 
     const postgresService = this.services.get(PostgresService);
@@ -117,7 +119,7 @@ class PostgresDatabase extends CustomResource<typeof specSchema> {
     const connectionError = await database.ping();
     if (connectionError) {
       console.error('Failed to connect', connectionError);
-      return;
+      throw new NotReadyError('FailedToConnectToDatabase');
     }
     await database.upsertRole({
       name: secret.user,

src/resources/istio/destination-rule/destination-rule.ts

@@ -15,12 +15,17 @@ class DestinationRule extends Resource<KubernetesObject & K8SDestinationRuleV1>
     super(options);
     const resourceService = this.services.get(ResourceService);
     this.#crd = resourceService.get(CRD, 'destinationrules.networking.istio.io');
+    this.#crd.on('changed', this.#handleChange);
   }
 
   public get hasCRD() {
     return this.#crd.exists;
   }
 
+  #handleChange = () => {
+    this.emit('changed', this.manifest);
+  };
+
   public set = async (manifest: KubernetesObject & K8SDestinationRuleV1) => {
     if (!this.hasCRD) {
       throw new NotReadyError('CRD is not installed');

src/resources/istio/gateway/gateway.ts

@@ -15,11 +15,11 @@ class Gateway extends Resource<KubernetesObject & K8SGatewayV1> {
     super(options);
     const resourceService = this.services.get(ResourceService);
     this.#crd = resourceService.get(CRD, 'gateways.networking.istio.io');
-    this.on('changed', this.#handleUpdate);
+    this.#crd.on('changed', this.#handleUpdate);
   }
 
   #handleUpdate = async () => {
-    this.emit('changed');
+    this.emit('changed', this.manifest);
   };
 
   public get hasCRD() {

src/resources/istio/virtual-service/virtual-service.ts

@@ -15,12 +15,17 @@ class VirtualService extends Resource<KubernetesObject & K8SVirtualServiceV1> {
     super(options);
     const resourceService = this.services.get(ResourceService);
     this.#crd = resourceService.get(CRD, 'virtualservices.networking.istio.io');
+    this.#crd.on('changed', this.#handleChange);
   }
 
   public get hasCRD() {
     return this.#crd.exists;
   }
 
+  #handleChange = () => {
+    this.emit('changed', this.manifest);
+  };
+
   public set = async (manifest: KubernetesObject & K8SVirtualServiceV1) => {
     if (!this.hasCRD) {
       throw new NotReadyError('CRD is not installed');

src/services/resources/resource/resource.custom.ts

@@ -1,11 +1,13 @@
 import { z, type ZodType } from 'zod';
-import { type KubernetesObject } from '@kubernetes/client-node';
+import { PatchStrategy, setHeaderOptions, type KubernetesObject } from '@kubernetes/client-node';
 
 import { Resource, type ResourceOptions } from './resource.ts';
 
 import { API_VERSION } from '#utils/consts.ts';
 import { CoalescingQueued } from '#utils/queues.ts';
 import { NotReadyError } from '#utils/errors.ts';
+import { K8sService } from '#services/k8s/k8s.ts';
+import { CronJob, CronTime } from 'cron';
 
 const customResourceStatusSchema = z.object({
   observedGeneration: z.number().optional(),
@@ -15,7 +17,7 @@ const customResourceStatusSchema = z.object({
         observedGeneration: z.number().optional(),
         type: z.string(),
         status: z.enum(['True', 'False', 'Unknown']),
-        lastTransitionTime: z.string().datetime(),
+        lastTransitionTime: z.string().datetime().optional(),
         resource: z.boolean().optional(),
         failed: z.boolean().optional(),
         syncing: z.boolean().optional(),
@@ -35,13 +37,14 @@ class CustomResource<TSpec extends ZodType> extends Resource<
   public static readonly status = customResourceStatusSchema;
 
   #reconcileQueue: CoalescingQueued<void>;
+  #cron: CronJob;
 
   constructor(options: CustomResourceOptions<TSpec>) {
     super(options);
     this.#reconcileQueue = new CoalescingQueued({
       action: async () => {
         try {
-          if (!this.exists || !this.manifest?.metadata?.deletionTimestamp) {
+          if (!this.exists || this.manifest?.metadata?.deletionTimestamp) {
             return;
           }
           this.services.log.debug('Reconciling', {
@@ -50,25 +53,63 @@ class CustomResource<TSpec extends ZodType> extends Resource<
             namespace: this.namespace,
             name: this.name,
           });
+          await this.markSeen();
           await this.reconcile?.();
+          await this.markReady();
         } catch (err) {
           if (err instanceof NotReadyError) {
-            console.error(err);
+            await this.markNotReady(err.reason, err.message);
+          } else if (err instanceof Error) {
+            await this.markNotReady('Failed', err.message);
           } else {
-            throw err;
+            await this.markNotReady('Failed', String(err));
           }
+          console.error(err);
         }
       },
     });
+    this.#cron = CronJob.from({
+      cronTime: '*/2 * * * *',
+      onTick: this.queueReconcile,
+      start: true,
+      runOnInit: true,
+    });
     this.on('changed', this.#handleUpdate);
   }
 
+  public get reconcileTime() {
+    return this.#cron.cronTime.toString();
+  }
+
+  public set reconcileTime(pattern: string) {
+    this.#cron.cronTime = new CronTime(pattern);
+  }
+
   public get isSeen() {
     return this.metadata?.generation === this.status?.observedGeneration;
   }
 
-  #handleUpdate = async () => {
-    if (this.isSeen) {
+  public get version() {
+    const [, version] = this.apiVersion.split('/');
+    return version;
+  }
+
+  public get group() {
+    const [group] = this.apiVersion.split('/');
+    return group;
+  }
+
+  public get scope() {
+    if (!('scope' in this.constructor) || typeof this.constructor.scope !== 'string') {
+      return;
+    }
+    return this.constructor.scope as 'Namespaced' | 'Cluster';
+  }
+
+  #handleUpdate = async (
+    previous?: KubernetesObject & { spec: z.infer<TSpec>; status?: z.infer<typeof customResourceStatusSchema> },
+  ) => {
+    if (this.isSeen && previous) {
       return;
     }
     return await this.queueReconcile();
@@ -88,9 +129,58 @@ class CustomResource<TSpec extends ZodType> extends Resource<
     });
   };
 
-  public patchStatus = async (status: Partial<z.infer<typeof customResourceStatusSchema>>) => {
-    this.patch({ status } as ExpectedAny);
+  public markNotReady = async (reason?: string, message?: string) => {
+    await this.patchStatus({
+      conditions: [
+        {
+          type: 'Ready',
+          status: 'False',
+          reason,
+          message,
+        },
+      ],
+    });
   };
+
+  public markReady = async () => {
+    await this.patchStatus({
+      conditions: [
+        {
+          type: 'Ready',
+          status: 'True',
+        },
+      ],
+    });
+  };
+
+  public patchStatus = (status: Partial<z.infer<typeof customResourceStatusSchema>>) =>
+    this.queue.add(async () => {
+      const k8sService = this.services.get(K8sService);
+      if (this.scope === 'Cluster') {
+        await k8sService.customObjectsApi.patchClusterCustomObjectStatus(
+          {
+            version: this.version,
+            group: this.group,
+            plural: this.plural,
+            name: this.name,
+            body: { status },
+          },
+          setHeaderOptions('Content-Type', PatchStrategy.MergePatch),
+        );
+      } else {
+        await k8sService.customObjectsApi.patchNamespacedCustomObjectStatus(
+          {
+            version: this.version,
+            group: this.group,
+            plural: this.plural,
+            name: this.name,
+            namespace: this.namespace || 'default',
+            body: { status },
+          },
+          setHeaderOptions('Content-Type', PatchStrategy.MergePatch),
+        );
+      }
+    });
 }
 
 export { CustomResource, type CustomResourceOptions };

src/services/resources/resource/resource.ts

@@ -20,11 +20,11 @@ type ResourceOptions<T extends KubernetesObject> = {
   manifest?: T;
 };
 
-type ResourceEvents = {
-  changed: () => void;
+type ResourceEvents<T extends KubernetesObject> = {
+  changed: (from?: T) => void;
 };
 
-class Resource<T extends KubernetesObject> extends EventEmitter<ResourceEvents> {
+class Resource<T extends KubernetesObject> extends EventEmitter<ResourceEvents<T>> {
   #manifest?: T;
   #queue: Queue;
   #options: ResourceOptions<T>;
@@ -36,6 +36,10 @@ class Resource<T extends KubernetesObject> extends EventEmitter<ResourceEvents>
     this.#queue = new Queue({ concurrency: 1 });
   }
 
+  protected get queue() {
+    return this.#queue;
+  }
+
   public get services() {
     return this.#options.services;
   }
@@ -48,8 +52,19 @@ class Resource<T extends KubernetesObject> extends EventEmitter<ResourceEvents>
     if (deepEqual(this.manifest, value)) {
       return;
     }
+    const previous = this.#manifest;
     this.#manifest = value;
-    this.emit('changed');
+    this.emit('changed', previous);
+  }
+
+  public get plural() {
+    if ('plural' in this.constructor && typeof this.constructor.plural === 'string') {
+      return this.constructor.plural;
+    }
+    if ('kind' in this.constructor && typeof this.constructor.kind === 'string') {
+      return this.constructor.kind.toLowerCase() + 's';
+    }
+    throw new Error('Unknown kind');
   }
 
   public get exists() {
@@ -123,11 +138,7 @@ class Resource<T extends KubernetesObject> extends EventEmitter<ResourceEvents>
   public patch = (patch: T) =>
     this.#queue.add(async () => {
       const { services } = this.#options;
-      services.log.debug(`Patching ${this.apiVersion}/${this.kind}/${this.namespace}/${this.name}`, {
-        spelector: this.selector,
-        current: this.manifest,
-        patch,
-      });
+      services.log.debug(`Patching ${this.apiVersion}/${this.kind}/${this.namespace}/${this.name}`);
       const k8s = services.get(K8sService);
       const body = {
         ...patch,

src/services/watchers/watchers.watcher.ts

@@ -47,9 +47,7 @@ class Watcher<T extends KubernetesObject> extends EventEmitter<WatcherEvents<T>>
     informer.on('update', this.#handleResource.bind(this, 'update'));
     informer.on('delete', this.#handleResource.bind(this, 'delete'));
     informer.on('error', (err) => {
-      if (!(err instanceof ApiException && err.code === 404)) {
-        console.log('Watcher failed, will retry in 3 seconds', path, err);
-      }
+      console.log('Watcher failed, will retry in 3 seconds', path, err);
       setTimeout(this.start, 3000);
     });
     return informer;