feat: add subscription

.gitignore

@@ -9,3 +9,7 @@
 .history
 .idea
 .vscode
+node_modules
+
+# MacBook
+.DS_Store

qlkube/.gitignore

@@ -1 +1,2 @@
 node_modules
+.env

qlkube/package.json

@@ -30,9 +30,13 @@
   },
   "dependencies": {
     "@graphql-tools/schema": "^7.1.4",
+    "@kubernetes/client-node": "^0.14.3",
+    "apollo-server": "2.24.0",
     "apollo-server-express": "^2.9.7",
     "compression": "^1.7.4",
+    "dotenv": "^9.0.2",
     "express": "^4.17.1",
+    "express-graphql": "^0.12.0",
     "got": "^9.6.0",
     "graphql": "^14.5.8",
     "graphql-tools": "^4.0.6",

qlkube/src/decorateBaseSchema.js

@@ -1,6 +1,7 @@
 const { gql } = require('apollo-server-core');
 const { extendSchema } = require('graphql/utilities');
 const { addResolversToSchema } = require('@graphql-tools/schema');
+const { capitalizeType } = require('./utils.js');
 
 /**
  *
@@ -12,63 +13,59 @@ const { addResolversToSchema } = require('@graphql-tools/schema');
  * @returns
  */
 
-function capitalizeType(name) {
-  return name[0].toUpperCase() + name.slice(1);
-}
-
-module.exports = {
-  decorateBaseSchema: function (
-    targetQuery,
-    extendedType,
-    baseSchema,
-    nameWrapper,
-    argsNeeded
-  ) {
-    if (!targetQuery) throw 'Parameter targetQuery cannot be empty!';
-    if (!extendedType) throw 'Parameter extendedType cannot be empty!';
-    if (!nameWrapper) throw 'Parameter nameWrapper cannot be empty!';
-    if (!argsNeeded) throw 'Parameter argsNeeded cannot be empty!';
-    if (!baseSchema) throw 'Parameter baseSchema cannot be empty!';
+function decorateBaseSchema(
+  targetQuery,
+  extendedType,
+  baseSchema,
+  nameWrapper,
+  argsNeeded
+) {
+  if (!targetQuery) throw 'Parameter targetQuery cannot be empty!';
+  if (!extendedType) throw 'Parameter extendedType cannot be empty!';
+  if (!nameWrapper) throw 'Parameter nameWrapper cannot be empty!';
+  if (!argsNeeded) throw 'Parameter argsNeeded cannot be empty!';
+  if (!baseSchema) throw 'Parameter baseSchema cannot be empty!';
 
-    if (baseSchema.getQueryType().getFields()[targetQuery] === undefined)
-      throw 'Parameter targetQuery not valid!';
-    const targetType = baseSchema.getQueryType().getFields()[targetQuery];
+  if (baseSchema.getQueryType().getFields()[targetQuery] === undefined)
+    throw 'Parameter targetQuery not valid!';
+  const targetType = baseSchema.getQueryType().getFields()[targetQuery];
 
-    if (!targetType) throw 'targetType fault!';
+  if (!targetType) throw 'targetType fault!';
 
-    let typeWrapper = capitalizeType(nameWrapper);
-    let typeTargetQuery = capitalizeType(targetQuery);
+  const typeWrapper = capitalizeType(nameWrapper);
+  const typeTargetQuery = capitalizeType(targetQuery);
 
-    const extension = gql`
-            extend type ${extendedType} {
-                ${nameWrapper}: ${typeWrapper}
-            }
-            type ${typeWrapper} {
-                ${targetQuery}: ${typeTargetQuery}
-            }
-        `;
-
-    const resolvers = {
-      [extendedType]: {
-        [nameWrapper]: (parent, args, context, info) => {
-          let newParent = {};
-          for (e of argsNeeded) {
-            if (parent[e] === undefined)
-              throw `Error: ${e} is not into the parent object!`;
-            newParent[e] = parent[e];
+  const extension = gql`
+          extend type ${extendedType} {
+              ${nameWrapper}: ${typeWrapper}
+          }
+          type ${typeWrapper} {
+              ${targetQuery}: ${typeTargetQuery}
           }
-          return newParent;
-        },
+      `;
+
+  const resolvers = {
+    [extendedType]: {
+      [nameWrapper]: (parent, args, context, info) => {
+        const newParent = {};
+        for (e of argsNeeded) {
+          if (parent[e] === undefined)
+            throw `Error: ${e} is not into the parent object!`;
+          newParent[e] = parent[e];
+        }
+        return newParent;
       },
-      [typeWrapper]: {
-        [targetQuery]: (parent, args, context, info) => {
-          return targetType.resolve(parent, parent, context, info);
-        },
+    },
+    [typeWrapper]: {
+      [targetQuery]: (parent, args, context, info) => {
+        return targetType.resolve(parent, parent, context, info);
       },
-    };
+    },
+  };
+
+  const extendedSchema = extendSchema(baseSchema, extension);
+  const newSchema = addResolversToSchema(extendedSchema, resolvers);
+  return newSchema;
+}
 
-    const extendedSchema = extendSchema(baseSchema, extension);
-    const newSchema = addResolversToSchema(extendedSchema, resolvers);
-    return newSchema;
-  },
-};
+module.exports = { decorateBaseSchema };

qlkube/src/decorateSubscription.js

@@ -0,0 +1,120 @@
+const { gql } = require('apollo-server-core');
+const { extendSchema } = require('graphql/utilities');
+const { addResolversToSchema } = require('@graphql-tools/schema');
+const { PubSub, withFilter } = require('apollo-server');
+const { capitalizeType } = require('./utils.js');
+
+const pubsub = new PubSub();
+
+function decorateEnum(baseSchema, enumName, values) {
+  if (!baseSchema) throw 'Parameter baseSchema cannot be empty!';
+  if (!enumName) throw 'Parameter enumName cannot be empty!';
+  if (!values) throw 'Parameter values cannot be empty!';
+
+  if (baseSchema._typeMap[enumName] !== undefined)
+    throw 'Enum type is already present in the schema!';
+
+  let enumType = `enum ${enumName} {`;
+  values.forEach(val => {
+    enumType += `
+      ${val}`;
+  });
+  enumType += `
+    }`;
+  const extension = gql`
+    ${enumType}
+  `;
+
+  const newSchema = extendSchema(baseSchema, extension);
+  return newSchema;
+}
+
+function decorateSubscription(baseSchema, targetType, enumType) {
+  if (!baseSchema) throw 'Parameter baseSchema cannot be empty!';
+  if (!targetType) throw 'Parameter targetType cannot be empty!';
+  if (!enumType) throw 'Parameter enumType cannot be empty!';
+
+  if (baseSchema.getQueryType().getFields()[targetType] === undefined)
+    throw 'Target type not found into the schema';
+
+  const subscriptionField = `${targetType}Update`;
+  const label = targetType;
+
+  const subType =
+    baseSchema._typeMap.Subscription === undefined
+      ? 'type Subscription'
+      : 'extend type Subscription';
+
+  const subscriptionType = capitalizeType(subscriptionField);
+  targetType = capitalizeType(targetType);
+
+  const extension = gql`
+  type  ${subscriptionType} {
+    updateType: ${enumType}
+    payload: ${targetType}
+  }
+    ${subType} {
+      ${subscriptionField}(name: String, namespace: String!): ${subscriptionType}
+    }
+  `;
+
+  const resolvers = {
+    Subscription: {
+      [subscriptionField]: {
+        subscribe: withFilter(
+          () => pubsub.asyncIterator([label]),
+          (payload, variables, info, context) => {
+            return (
+              payload.apiObj.metadata.namespace === variables.namespace &&
+              (variables.name === undefined ||
+                payload.apiObj.metadata.name === variables.name)
+            );
+          }
+        ),
+        resolve: async (payload, args, context, info) => {
+          return payload;
+        },
+      },
+    },
+    [subscriptionType]: {
+      updateType: (payload, args, context, info) => {
+        return payload.type;
+      },
+      payload: (payload, args, context, info) => {
+        return payload.apiObj;
+      },
+    },
+  };
+
+  const extendedSchema = extendSchema(baseSchema, extension);
+  const newSchema = addResolversToSchema(extendedSchema, resolvers);
+  newSchema._subscriptionType = newSchema._typeMap.Subscription;
+  return newSchema;
+}
+
+function setupSubscriptions(subscriptions, schema) {
+  if (!subscriptions) throw 'Parameter subscriptions cannot be empty!';
+  if (!schema) throw 'Parameter schema cannot be empty!';
+
+  let newSchema = decorateEnum(schema, 'UpdateType', [
+    'ADDED',
+    'MODIFIED',
+    'DELETED',
+  ]);
+
+  subscriptions.forEach(e => {
+    newSchema = decorateSubscription(newSchema, e.type, 'UpdateType');
+  });
+
+  return newSchema;
+}
+
+function publishEvent(label, value) {
+  pubsub.publish(label, value);
+}
+
+module.exports = {
+  decorateSubscription,
+  setupSubscriptions,
+  publishEvent,
+};

qlkube/src/index.js

@@ -1,11 +1,18 @@
 const fs = require('fs').promises;
+const { createServer } = require('http');
 const express = require('express');
 const { ApolloServer } = require('apollo-server-express');
 const compression = require('compression');
 const { createSchema } = require('./schema');
+const { kwatch } = require('./watch.js');
+const { setupSubscriptions } = require('./decorateSubscription.js');
+const { subscriptions } = require('./subscriptions.js');
 const getOpenApiSpec = require('./oas');
 const { printSchema } = require('graphql');
 const logger = require('pino')({ useLevelLabels: true });
+const dotenv = require('dotenv');
+
+dotenv.config();
 
 main().catch(e =>
   logger.error({ error: e.stack }, 'failed to start qlkube server')
@@ -25,16 +32,37 @@ async function main() {
     : '';
 
   const oas = await getOpenApiSpec(kubeApiUrl, token);
-  const schema = await createSchema(oas, kubeApiUrl, token);
+  let schema = await createSchema(oas, kubeApiUrl, token);
+
+  try {
+    schema = setupSubscriptions(subscriptions, schema);
+  } catch (e) {
+    console.error(e);
+    process.exit(1);
+  }
 
   const server = new ApolloServer({
     schema,
-    context: ({ req }) => {
-      if (req.headers.authorization && req.headers.authorization.length > 0) {
-        const strs = req.headers.authorization.split(' ');
-        var user = {};
-        user.token = strs[1];
-        return user;
+    subscriptions: {
+      path: '/subscription',
+      onConnect: (connectionParams, webSocket, context) => {
+        console.log('Connected!');
+      },
+      onDisconnect: (webSocket, context) => {
+        console.log('Disconnected!');
+      },
+    },
+
+    context: ({ req, connection }) => {
+      if (connection) {
+        return {};
+      } else {
+        if (req.headers.authorization && req.headers.authorization.length > 0) {
+          const strs = req.headers.authorization.split(' ');
+          var user = {};
+          user.token = strs[1];
+          return user;
+        }
       }
     },
   });
@@ -52,10 +80,26 @@ async function main() {
     app,
     path: '/',
   });
-  app.listen({ port: 8080 }, () =>
-    logger.info(
-      { url: `http://localhost:8080${server.graphqlPath}` },
-      '🚀 Server ready'
-    )
-  );
+  const httpServer = createServer(app);
+  server.installSubscriptionHandlers(httpServer);
+
+  const PORT = process.env.CROWNLABS_QLKUBE_PORT || 8080;
+
+  httpServer.listen({ port: PORT }, () => {
+    console.log(
+      `🚀 Server ready at http://localhost:${PORT}${server.graphqlPath}`
+    );
+    console.log(
+      `🚀 Subscriptions ready at ws://localhost:${PORT}${server.subscriptionsPath}`
+    );
+  });
+
+  try {
+    subscriptions.forEach(sub => {
+      kwatch(sub.resource, sub.type);
+    });
+  } catch (e) {
+    console.error(e);
+    process.exit(1);
+  }
 }

qlkube/src/schema.js

@@ -7,14 +7,14 @@ exports.createSchema = async (oas, kubeApiUrl, token) => {
   let baseSchema = await oasToGraphQlSchema(oas, kubeApiUrl, token);
   let allSchema = decorateSchema(baseSchema);
   try {
-    let schemaWithInstanceTemplate = decorateBaseSchema(
+    const schemaWithInstanceTemplate = decorateBaseSchema(
       'itPolitoCrownlabsV1alpha2Template',
       'TemplateCrownlabsPolitoItTemplateRef',
       allSchema,
       'templateWrapper',
       ['name', 'namespace']
     );
-    let schemaWithInstanceTenant = decorateBaseSchema(
+    const schemaWithInstanceTenant = decorateBaseSchema(
       'itPolitoCrownlabsV1alpha1Tenant',
       'TenantCrownlabsPolitoItTenantRef',
       schemaWithInstanceTemplate,

qlkube/src/subscriptions.js

@@ -0,0 +1,6 @@
+const subscriptions = [
+  { resource: '/api/v1/pods', type: 'ioK8sApiCoreV1Pod' },
+  { resource: '/api/v1/nodes', type: 'ioK8sApiCoreV1Node' },
+];
+
+module.exports = { subscriptions };

qlkube/src/utils.js

@@ -0,0 +1,5 @@
+function capitalizeType(name) {
+  return name[0].toUpperCase() + name.slice(1);
+}
+
+module.exports = { capitalizeType };