Add support for metrics-server

sebt3 · Mar 27, 2024 · 0821d51 · 0821d51
1 parent 0d4855e
commit 0821d51
Show file tree

Hide file tree

Showing 26 changed files with 3,306 additions and 1,064 deletions.
diff --git a/back/main.ts b/back/main.ts
@@ -2,6 +2,9 @@ import { ApolloServer } from '@apollo/server';
 import { expressMiddleware } from '@apollo/server/express4';
 import { ApolloServerPluginLandingPageLocalDefault, ApolloServerPluginLandingPageProductionDefault } from '@apollo/server/plugin/landingPage/default';
 import { ApolloServerPluginDrainHttpServer } from '@apollo/server/plugin/drainHttpServer';
+import { makeExecutableSchema } from '@graphql-tools/schema';
+import { WebSocketServer } from 'ws';
+import { useServer } from 'graphql-ws/lib/use/ws';
 import express from 'express';
 import http from 'http';
 import cors from 'cors';
@@ -51,7 +54,21 @@ interface MyContext {
 }
 export const app = express();
 export const httpServer = http.createServer(app);
-const apolloPlugins = [ApolloServerPluginDrainHttpServer({ httpServer })]
+/*const wsServer = new WebSocketServer({
+  server: httpServer,
+  path: '/subscriptions',
+});
+const schema = makeExecutableSchema({ typeDefs, resolvers });
+const serverCleanup = useServer({ schema }, wsServer);*/
+const apolloPlugins = [ApolloServerPluginDrainHttpServer({ httpServer }),/*{
+  async serverWillStart() {
+    return {
+      async drainServer() {
+        await serverCleanup.dispose();
+      },
+    };
+  },
+}*/]
 if (gramoConfig.enableGraphQLClient||process.env.NODE_ENV !== 'production')
   apolloPlugins.push(ApolloServerPluginLandingPageLocalDefault());
 else

diff --git a/back/resolvers/core/Container.ts b/back/resolvers/core/Container.ts
@@ -1,9 +1,12 @@
 import k8s from '@kubernetes/client-node';
-import { lists as listNamespace } from '../k8s/Namespace.js';
-import {kc, cache, applyFilter, applyFieldSelection, getByPath, getMeta } from '../k8slibs.js';
-import { knowledge } from '../knowledge.js'
+//import { lists as listNamespace } from '../k8s/Namespace.js';
+import {kc, cache /*, applyFilter, applyFieldSelection, getByPath, getMeta*/ } from '../k8slibs.js';
+//import { knowledge } from '../knowledge.js'
+//import { LogPubSub } from '../../pubsub/logpubsub.js'
 const k8sApi = kc.makeApiClient(k8s.CoreV1Api);
-const short2plural = (short:string) => short.toLowerCase()+'s'
+const log = new k8s.Log(kc);
+
+//const short2plural = (short:string) => short.toLowerCase()+'s'
 export const mutations = {
 };
 export const lists = {

diff --git a/back/resolvers/core/NodeMetrics.ts b/back/resolvers/core/NodeMetrics.ts
@@ -0,0 +1,54 @@
+import {cache, rawQuery, applyFilter, applyFieldSelection} from '../k8slibs.js';
+const metricMaxLength = 500;
+export const mutations = {
+};
+export const lists = {
+  coreNodeMetrics: async (parent, args: object) => {
+    const ret:{name:string,data:{timestamp: string, window: string, cpu: string, memory: string}[]}[] = cache.get('coreNodeMetrics') || []
+    const lst:Array<object>|undefined = cache.get('coreNodeMetricsInternal')
+    if (lst==undefined) {
+      try {
+        const parsed = await rawQuery('/apis/metrics.k8s.io/v1beta1/nodes') as {items:{metadata:{name:string}, window:string, timestamp:string, usage:{cpu:string,memory:string}}[]}
+        if (!Array.isArray(parsed.items)) throw new Error(`No data`);
+        parsed.items.forEach(i=>{
+            const line = {
+              timestamp: i.timestamp,
+              window: i.window,
+              cpu: i.usage.cpu,
+              memory: i.usage.memory
+            }
+            if (ret.filter(node=>node.name==i.metadata.name).length>0) {
+              if (ret.filter(node=>node.name==i.metadata.name)[0].data.filter(d=>d.timestamp==i.timestamp).length<1)
+                ret.filter(node=>node.name==i.metadata.name)[0].data.push(line)
+            } else {
+              ret.push({
+                name: i.metadata.name,
+                data: [line]
+              })
+            }
+            if (ret.filter(node=>node.name==i.metadata.name)[0].data.length>500)
+                ret.filter(node=>node.name==i.metadata.name)[0].data.pop();
+        });
+        cache.set('coreNodeMetricsInternal', [], 2);
+        cache.set('coreNodeMetrics', ret, 2*metricMaxLength);
+        return ret
+      } catch (err) {
+        if (typeof err === 'object' && (err as object)['body'] !=undefined) {
+          if ((err as object)['body']['reason']!='Forbidden') {
+            console.error('error', (err as object)['body']);
+          } else {
+            cache.set('coreNodeMetricsInternal', [], 2);
+          }
+        } else {console.error('error', err)}
+        return []
+      }
+    }
+    if (lst===undefined) return []
+    return lst.filter(o=>applyFilter(o,args)).map(o=>applyFieldSelection(o,args))
+  }
+};
+export const queries = {
+    coreNodeMetrics: lists.coreNodeMetrics,
+};
+export const resolvers = {
+};
diff --git a/back/resolvers/core/PodMetrics.ts b/back/resolvers/core/PodMetrics.ts
@@ -0,0 +1,50 @@
+import {cache, rawQuery, applyFilter, applyFieldSelection} from '../k8slibs.js';
+const metricMaxLength = 500;
+export const mutations = {
+};
+export const lists = {
+  corePodMetrics: async (parent, args: object) => {
+    const ret:{pod_name:string, name:string,namespace:string,data:{timestamp: string, window: string, cpu: string, memory:string}[]}[] = cache.get('corePodMetrics') || []
+    const lst:Array<object>|undefined = cache.get('corePodMetricsInternal')
+    if (lst==undefined) {
+      try {
+        const parsed = await rawQuery('/apis/metrics.k8s.io/v1beta1/pods') as {items:{metadata:{name:string,namespace:string}, containers:object[],window:string, timestamp:string}[]}
+        if (!Array.isArray(parsed.items)) throw new Error(`No data`);
+        parsed.items.map(i=>i.containers.map(c=>{return {pod_name:i.metadata.name, namespace:i.metadata.namespace, name: c['name'], data: {timestamp: i.timestamp, window: i.window, cpu: c['usage']['cpu'], memory: c['usage']['memory']}}})).flat().forEach(i=>{
+            if (ret.filter(pod=>pod.pod_name==i.pod_name&&pod.name==i.name&&pod.namespace==i.namespace).length>0) {
+              if (ret.filter(pod=>pod.pod_name==i.pod_name&&pod.name==i.name&&pod.namespace==i.namespace)[0].data.filter(d=>d.timestamp==i.data.timestamp).length<1)
+                ret.filter(pod=>pod.pod_name==i.pod_name&&pod.name==i.name&&pod.namespace==i.namespace)[0].data.push(i.data)
+            } else {
+              ret.push({
+                name: i.name,
+                pod_name: i.pod_name,
+                namespace: i.namespace,
+                data: [i.data]
+              })
+            }
+            if (ret.filter(pod=>pod.pod_name==i.pod_name&&pod.name==i.name&&pod.namespace==i.namespace)[0].data.length>500)
+                ret.filter(pod=>pod.pod_name==i.pod_name&&pod.name==i.name&&pod.namespace==i.namespace)[0].data.pop();
+        });
+        cache.set('corePodMetricsInternal', [], 2);
+        cache.set('corePodMetrics', ret, 2*metricMaxLength);
+        return ret
+      } catch (err) {
+        if (typeof err === 'object' && (err as object)['body'] !=undefined) {
+          if ((err as object)['body']['reason']!='Forbidden') {
+            console.error('error', (err as object)['body']);
+          } else {
+            cache.set('corePodMetricsInternal', [], 2);
+          }
+        } else {console.error('error', err)}
+        return []
+      }
+    }
+    if (lst===undefined) return []
+    return lst.filter(o=>applyFilter(o,args)).map(o=>applyFieldSelection(o,args))
+  }
+};
+export const queries = {
+    corePodMetrics: lists.corePodMetrics,
+};
+export const resolvers = {
+};
diff --git a/back/resolvers/k8slibs.ts b/back/resolvers/k8slibs.ts
@@ -1,4 +1,5 @@
 import k8s from '@kubernetes/client-node';
+import request from 'request';
 import NodeCache from 'node-cache';
 export const kc = new k8s.KubeConfig();
 export const cache = new NodeCache({ stdTTL: 2, useClones: false, deleteOnExpire: true, checkperiod: 60 });
@@ -21,6 +22,17 @@ function deleteByPath(obj,path) {
         delete parent[path.split("/").slice(-1)[0]];
 }
 
+const opts = {} as request.Options;
+kc.applyToRequest(opts);
+export const rawQuery = async (path:string) => {
+    return new Promise((resolve,reject) => request.get(kc.getCurrentCluster().server + path, opts, (error, response, body) => {
+        if (error) reject(error);
+        if (response === undefined || response==null) { reject('no response from API'); return }
+        if (response.statusCode != 200) { reject(`Query failed, ${response.statusCode}`); return }
+        resolve(JSON.parse(body))
+    }));
+}
+
 function addByPath(target,path,data) {
     path.split("/").slice(0,-1).reduce((res,cur) => {if (res[cur]==undefined) res[cur]={};return res[cur]},target)[path.split("/").slice(-1)[0]] = data
 }

diff --git a/back/schema/core.graphql b/back/schema/core.graphql
@@ -154,17 +154,36 @@ type coreContainer {
   status: JSONObject
   getcoreLog(params: queryParameters): coreLog
 }
+type coreMetricsData {
+  timestamp: String!
+  window: String!
+  cpu: String!
+  memory: String!
+}
+type corePodMetrics {
+  namespace: String!
+  pod_name: String!
+  name: String!
+  data: [coreMetricsData]
+}
+type coreNodeMetrics {
+  name: String!
+  data: [coreMetricsData]
+}
 type Query {
   gramoConfig: GramoConfig
   coreEvent(params: queryParameters): [coreEvent]
   vynilCategory(params: queryParameters): [vynilCategory]
   vynilPackage(params: queryParameters): [vynilPackage]
 }
 type Mutation {
-    namespacedCrdObjectCreate(group: String!, version: String!, plural: String!, kind: String!, metadata: metadataInput!, spec: JSONObject): coreCrdObject
-    namespacedCrdObjectPatch(group: String!, version: String!, plural: String!, kind: String!, metadata: metadataInput!, spec: JSONObject): coreCrdObject
-    namespacedCrdObjectDelete(group: String!, version: String!, plural: String!, namespace: String!, name: String!): coreCrdObject
-    clusteredCrdObjectCreate(group: String!, version: String!, plural: String!, kind: String!, metadata: metadataInput!, spec: JSONObject): coreCrdObject
-    clusteredCrdObjectPatch(group: String!, version: String!, plural: String!, kind: String!, metadata: metadataInput!, spec: JSONObject): coreCrdObject
-    clusteredCrdObjectDelete(group: String!, version: String!, plural: String!, name: String!): coreCrdObject
-}
+  namespacedCrdObjectCreate(group: String!, version: String!, plural: String!, kind: String!, metadata: metadataInput!, spec: JSONObject): coreCrdObject
+  namespacedCrdObjectPatch(group: String!, version: String!, plural: String!, kind: String!, metadata: metadataInput!, spec: JSONObject): coreCrdObject
+  namespacedCrdObjectDelete(group: String!, version: String!, plural: String!, namespace: String!, name: String!): coreCrdObject
+  clusteredCrdObjectCreate(group: String!, version: String!, plural: String!, kind: String!, metadata: metadataInput!, spec: JSONObject): coreCrdObject
+  clusteredCrdObjectPatch(group: String!, version: String!, plural: String!, kind: String!, metadata: metadataInput!, spec: JSONObject): coreCrdObject
+  clusteredCrdObjectDelete(group: String!, version: String!, plural: String!, name: String!): coreCrdObject
+}
+type Subscription {
+  onLog(namespace: String!,pod_name: String!,name: String!): coreLog
+}