Merge pull request #10 from ganochenkodg/leader-election-refactoring

0.2.3

README.md

@@ -9,7 +9,7 @@ Kubernetes operator for the [Qdrant](https://github.com/qdrant/qdrant) Vector Da
 
 The operator provides the following functionality:
 
-- Creation of single-node and multi-node Qdrant clusters (distributed mode is automatically enabled when using multiple replicas).
+- Creation of single-node and multi-node Qdrant clusters, cluster scaling.
 - Authentication support in the cluster using API keys with read-write and read-only permissions. Auto-generated and user-provided keys are supported.
 - TLS Encryption support for connections, both for client and peer-to-peer communication. Auto-generated and user-provided certificates are supported.
 - Support for setting custom Qdrant parameters.

changelog.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## qdrant-operator-0.2.3 (2024-01-18)
+
+- Fixed Qdrant cluster scaling 1 => N replicas
+- Added background checker for lease ownership
+- Bugfixes in the setStatus function
+
 ## qdrant-operator-0.2.2 (2024-01-17)
 
 Small bugfixes.

deploy/operator.yaml

@@ -78,7 +78,7 @@ spec:
       terminationGracePeriodSeconds: 10
       containers:
       - name: operator
-        image: docker.io/dganochenko/qdrant-operator:0.2.2
+        image: docker.io/dganochenko/qdrant-operator:0.2.3
         imagePullPolicy: Always
         # debug
         # command: ["/bin/sh"]

src/index.js

@@ -24,9 +24,9 @@ const lock = new K8SLock({
   lockLeaserId: process.env.POD_NAME,
   waitUntilLock: true,
   createLeaseIfNotExist: true,
-  leaseDurationInSeconds: 15,
+  leaseDurationInSeconds: 10,
   refreshLockInterval: 500,
-  lockTryInterval: 3000
+  lockTryInterval: 5000
 });
 
 // set debug mode, false by default
@@ -40,8 +40,6 @@ var clusterWatch = '';
 var collectionWatch = '';
 var clusterWatchStart = true;
 var collectionWatchStart = true;
-var lockId = '';
-var lockInfo = '';
 // load KubeConfig
 const kc = new k8s.KubeConfig();
 kc.loadFromDefault();
@@ -51,15 +49,11 @@ const k8sCoreApi = kc.makeApiClient(k8s.CoreV1Api);
 const k8sCustomApi = kc.makeApiClient(k8s.CustomObjectsApi);
 const k8sPolicyApi = kc.makeApiClient(k8s.PolicyV1Api);
 const k8sAppsApi = kc.makeApiClient(k8s.AppsV1Api);
+const k8sCoordinationApi = kc.makeApiClient(k8s.CoordinationV1Api);
 const watch = new k8s.Watch(kc);
 
 // react on QdrantClusters events
 const onEventCluster = async (phase, apiObj) => {
-  // leader status was lost
-  if (!lockInfo.isLocking) {
-    log('Leader status was lost, restarting...');
-    process.exit(1);
-  }
   // ignore MODIFIED on status changes
   if (settingStatus.has(apiObj.metadata.name)) {
     return;
@@ -86,11 +80,6 @@ const onEventCluster = async (phase, apiObj) => {
 
 // react on QdrantCollections events
 const onEventCollection = async (phase, apiObj) => {
-  // leader status was lost
-  if (!lockInfo.isLocking) {
-    log('Leader status was lost, restarting...');
-    process.exit(1);
-  }
   // ignore duplicated event on watch reconnections
   if (lastCollectionResourceVersion == apiObj.metadata.resourceVersion) {
     return;
@@ -197,7 +186,7 @@ const setStatus = async (apiObj, k8sCustomApi, status) => {
     log(err);
   }
   // job is done, remove this resource from the map
-  settingStatus.delete(name);
+  setTimeout(() => settingStatus.delete(name), 300);
 };
 
 // update the version of last caught cluster
@@ -215,6 +204,24 @@ const updateResourceVersion = async (apiObj, k8sCustomApi) => {
   lastClusterResourceVersion = resCurrent.metadata.resourceVersion;
 };
 
+// check the current leader
+const isLeader = async () => {
+  const namespace = process.env.POD_NAMESPACE;
+  try {
+    const res = await k8sCoordinationApi.readNamespacedLease(
+      'qdrant-operator',
+      namespace
+    );
+    // leader status was lost
+    if (res.body.spec.holderIdentity !== process.env.POD_NAME) {
+      log('Leader status was lost, restarting...');
+      process.exit(1);
+    }
+  } catch (err) {
+    log(err);
+  }
+};
+
 // check the cluster readiness
 const waitForClusterReadiness = (apiObj, k8sAppsApi, k8sCustomApi) => {
   // start background periodic job
@@ -289,9 +296,10 @@ const main = async () => {
   log(
     `Status of "${process.env.POD_NAME}": FOLLOWER. Trying to get leader status...`
   );
-  lockInfo = await lock.startLocking();
-  lockId = lockInfo.lockId;
+  const lockInfo = await lock.startLocking();
   log(`Status of "${process.env.POD_NAME}": LEADER.`);
+  // start checking lease ownership in background
+  setInterval(() => isLeader(), 10000);
   // start watching events only after taking ownership of the lease
   await watchResource();
 };
@@ -311,7 +319,7 @@ if (debugMode == 'true') {
 
 // got SIGTERM - stop locking and exit
 process.on('SIGTERM', async () => {
-  await lock.stopLocking(lockId);
+  await lock.stopLocking();
   log('Stopping gracefully...');
   process.exit(0);
 });

src/package.json

@@ -1,6 +1,6 @@
 {
   "name": "qdrant-operator",
-  "version": "0.2.2",
+  "version": "0.2.3",
   "description": "Kubernetes operator for Qdrant",
   "main": "index.js",
   "type": "module",

src/templates/statefulset.jsr

@@ -37,10 +37,8 @@ spec:
             - name: QDRANT__CLUSTER__P2P__ENABLE_TLS
               value: "true"
           {{/if}}
-          {{if spec.replicas > 1}}
             - name: QDRANT__CLUSTER__ENABLED
               value: "true"
-          {{/if}}
           command: ["/bin/bash", "-c"]
           args: ["./config/initialize.sh"]
           ports: