Extract hosts/pods/containers/services assets

constants.ts

@@ -1,8 +1,10 @@
 export const LOGS_INDICES = 'logs-*,filebeat-*';
-export const APM_INDICES = 'traces-*,apm*,metrics-apm*';
+export const APM_INDICES = 'traces-*,apm*,metrics-apm*,logs-apm*';
+export const METRICS_INDICES = 'metrics-*,metricbeat-*';
 
 export const REMOTE_LOGS_INDICES = 'remote_cluster:logs-*,remote_cluster:filebeat-*';
 export const REMOTE_APM_INDICES = 'remote_cluster:traces-*,remote_cluster:apm*,remote_cluster:metrics-apm*';
+export const REMOTE_METRICS_INDICES = 'remote_cluster:metrics-*,remote_cluster:metricbeat-*';
 
 export function getLogsIndices() {
   if (process.env.ES_IS_CCS === "true") {
@@ -18,4 +20,12 @@ export function getApmIndices() {
   } else {
     return APM_INDICES;
   }
-}
+}
+
+export function getMetricsIndices() {
+  if (process.env.ES_IS_CCS === "true") {
+    return REMOTE_METRICS_INDICES;
+  } else {
+    return METRICS_INDICES;
+  }
+}

lib/assets_index_template.ts

@@ -4,6 +4,7 @@ export const assetsIndexTemplateConfig: IndicesPutIndexTemplateRequest = {
   name: 'assets',
   index_patterns: ['assets*'],
   priority: 100,
+  data_stream: {},
   template: {
     settings: {},
     mappings: {

lib/collectContainers.ts

@@ -0,0 +1,73 @@
+import { Client } from "@elastic/elasticsearch";
+import { getApmIndices, getLogsIndices, getMetricsIndices } from "../constants";
+import { SimpleAsset } from "../types";
+
+interface CollectContainers {
+  containers: SimpleAsset[];
+}
+
+export async function collectContainers({ esClient }: { esClient: Client }) {
+  const dsl = {
+    index: [getLogsIndices(), getApmIndices(), getMetricsIndices()],
+    size: 1000,
+    collapse: {
+      field: 'container.id'
+    },
+    sort: [
+      { '_score': 'desc' },
+      { '@timestamp': 'desc' }
+    ],
+    _source: false,
+    fields: [
+      'kubernetes.*',
+      'cloud.provider',
+      'orchestrator.cluster.name',
+      'host.name',
+      'host.hostname'
+    ],
+    query: {
+      bool: {
+        filter: [
+          {
+            range: {
+              '@timestamp': {
+                gte: 'now-1h'
+              }
+            }
+          }
+        ],
+        should: [
+          { exists: { field: 'kubernetes.container.id' } },
+          { exists: { field: 'kubernetes.pod.uid' } },
+          { exists: { field: 'host.hostname' } },
+        ]
+      }
+    }
+  };
+
+  const esResponse = await esClient.search(dsl);
+
+  const docs = esResponse.hits.hits.reduce<CollectContainers>((acc, hit) => {
+    const { fields = {} } = hit;
+    const containerId = fields['container.id'];
+    const podUid = fields['kubernetes.pod.uid'];
+    const nodeName = fields['kubernetes.node.name'];
+
+    const parentEan = podUid ? `pod:${podUid}` : `host:${fields['host.hostname']}`;
+
+    const container: SimpleAsset = {
+      '@timestamp': new Date(),
+      'asset.kind': 'container',
+      'asset.id': containerId,
+      'asset.ean': `container:${containerId}`,
+      'asset.parents': [parentEan],
+    };
+
+    acc.containers.push(container);
+
+    return acc;
+  }, { containers: [] });
+
+  return docs.containers;
+}
+

lib/collectHosts.ts

@@ -0,0 +1,98 @@
+import { Client } from "@elastic/elasticsearch";
+import { getApmIndices, getLogsIndices, getMetricsIndices } from "../constants";
+import { SimpleAsset } from "../types";
+
+interface CollectHosts {
+  hosts: SimpleAsset[];
+}
+
+export async function collectHosts({ esClient }: { esClient: Client }): Promise<SimpleAsset[]> {
+  const dsl = {
+    index: [getMetricsIndices(), getLogsIndices(), getApmIndices()],
+    size: 1000,
+    collapse: { field: 'host.hostname' },
+    sort: [
+      { "_score": "desc" },
+      { "@timestamp": "desc" }
+    ],
+    _source: false,
+    fields: [
+      '@timestamp',
+      'cloud.*',
+      'container.*',
+      'host.hostname',
+      'kubernetes.*',
+      'orchestrator.cluster.name',
+    ],
+    query: {
+      bool: {
+        filter: [
+          {
+            range: {
+              '@timestamp': {
+                gte: 'now-1h'
+              }
+            }
+          }
+        ],
+        must: [
+          { exists: { field: 'host.hostname' } },
+        ],
+        should: [
+          { exists: { field: 'kubernetes.node.name' } },
+          { exists: { field: 'kubernetes.pod.uid' } },
+          { exists: { field: 'container.id' } }
+        ]
+      }
+    }
+  };
+
+  const esResponse = await esClient.search(dsl);
+
+  const assets = esResponse.hits.hits.reduce<CollectHosts>((acc, hit) => {
+    const { fields = {} } = hit;
+    const hostName = fields['host.hostname'];
+    const k8sNode = fields['kubernetes.node.name'];
+    const k8sPod = fields['kubernetes.pod.uid'];
+
+    const hostEan = `host:${k8sNode || hostName}`;
+
+    const host: SimpleAsset = {
+      '@timestamp': new Date(),
+      'asset.kind': 'host',
+      'asset.id': k8sNode || hostName,
+      'asset.name': k8sNode || hostName,
+      'asset.ean': hostEan,
+    };
+
+    if (fields['cloud.provider']) {
+      host['cloud.provider'] = fields['cloud.provider'];
+    }
+
+    if (fields['cloud.instance.id']) {
+      host['cloud.instance.id'] = fields['cloud.instance.id'];
+    }
+
+    if (fields['cloud.service.name']) {
+      host['cloud.service.name'] = fields['cloud.service.name'];
+    }
+
+    if (fields['cloud.region']) {
+      host['cloud.region'] = fields['cloud.region'];
+    }
+
+    if (fields['orchestrator.cluster.name']) {
+      host['orchestrator.cluster.name'] = fields['orchestrator.cluster.name'];
+    }
+
+    if (k8sPod) {
+      host['asset.children'] = [`pod:${k8sPod}`];
+    }
+
+    acc.hosts.push(host);
+
+    return acc;
+  }, { hosts: [] });
+
+  return assets.hosts;
+}

lib/collectPods.ts

@@ -1,32 +1,24 @@
 import { Client } from "@elastic/elasticsearch";
-import { getApmIndices, getLogsIndices } from "../constants";
+import { getApmIndices, getLogsIndices, getMetricsIndices } from "../constants";
 import { SimpleAsset } from "../types";
 
-interface CollectPodsAndNodes {
-  pods: SimpleAsset<'k8s.pod'>[];
-  nodes: SimpleAsset<'k8s.node'>[];
+interface CollectPods {
+  pods: SimpleAsset[];
 }
 
 export async function collectPods({ esClient }: { esClient: Client }) {
-  // STEP ONE: Query pods that reference their k8s nodes
   const dsl = {
-    index: [getLogsIndices(), getApmIndices()],
+    index: [getLogsIndices(), getApmIndices(), getMetricsIndices()],
     size: 1000,
     collapse: {
       field: 'kubernetes.pod.uid'
     },
     sort: [
-      {
-        "@timestamp": "desc" // TODO: Switch to ASC with a hard-coded "one hour ago" value, then use "search_after" to process all results?
-      }
+      { '@timestamp': 'desc' }
     ],
     _source: false,
     fields: [
-      'kubernetes.pod.uid',
-      'kubneretes.pod.name',
-      'kubernetes.node.id',
-      'kubernetes.node.name',
-      'kubernetes.namespace',
+      'kubernetes.*',
       'cloud.provider',
       'orchestrator.cluster.name',
       'host.name',
@@ -44,37 +36,27 @@ export async function collectPods({ esClient }: { esClient: Client }) {
           }
         ],
         must: [
-          {
-            exists: {
-              field: 'kubernetes.pod.uid'
-            }
-          },
-          {
-            exists: {
-              field: 'kubernetes.node.name'
-            }
-          }
+          { exists: { field: 'kubernetes.pod.uid' } },
+          { exists: { field: 'kubernetes.node.name' } }
         ]
       }
     }
   };
 
-  console.log(JSON.stringify(dsl));
   const esResponse = await esClient.search(dsl);
 
-  // STEP TWO: Loop over collected pod documents and create a pod asset doc AND a node asset doc for each
-  const docs = esResponse.hits.hits.reduce<CollectPodsAndNodes>((acc, hit) => {
+  const docs = esResponse.hits.hits.reduce<CollectPods>((acc, hit) => {
     const { fields = {} } = hit;
     const podUid = fields['kubernetes.pod.uid'];
     const nodeName = fields['kubernetes.node.name'];
     const clusterName = fields['orchestrator.cluster.name'];
 
-    const pod: SimpleAsset<'k8s.pod'> = {
+    const pod: SimpleAsset = {
       '@timestamp': new Date(),
-      'asset.type': 'k8s.pod',
+      'asset.kind': 'pod',
       'asset.id': podUid,
-      'asset.ean': `k8s.pod:${podUid}`,
-      'asset.parents': [`k8s.node:${nodeName}`]
+      'asset.ean': `pod:${podUid}`,
+      'asset.parents': [`host:${nodeName}`]
     };
 
     if (fields['cloud.provider']) {
@@ -87,32 +69,8 @@ export async function collectPods({ esClient }: { esClient: Client }) {
 
     acc.pods.push(pod);
 
-    const foundNode = acc.nodes.find((collectedNode) => collectedNode['asset.ean'] === `k8s.node:${nodeName}`);
-
-    if (foundNode) {
-      if (foundNode['asset.children']) {
-        foundNode['asset.children'].push(`k8s.pod:${podUid}`);
-      } else {
-        foundNode['asset.children'] = [`k8s.pod:${podUid}`];
-      }
-    } else {
-      const node: SimpleAsset<'k8s.node'> = {
-        '@timestamp': new Date(),
-        'asset.type': 'k8s.node',
-        'asset.id': nodeName,
-        'asset.ean': `k8s.node:${nodeName}`,
-        'asset.children': [`k8s.pod:${podUid}`]
-      };
-
-      if (clusterName) {
-        node['asset.parents'] = [`k8s.cluster:${clusterName}`];
-      }
-
-      acc.nodes.push(node);
-    }
-
     return acc;
-  }, { pods: [], nodes: [] });
+  }, { pods: [] });
 
-  return { esResponse, docs };
-}
+  return docs.pods;
+}