Spike: Data Source Catalog for Detection Engineering

package.json

@@ -545,6 +545,7 @@
     "@kbn/data-search-plugin": "link:src/platform/test/plugin_functional/plugins/data_search",
     "@kbn/data-service": "link:src/platform/packages/shared/kbn-data-service",
     "@kbn/data-service-server": "link:src/platform/packages/shared/kbn-data-service-server",
+    "@kbn/data-source-catalog": "link:x-pack/platform/packages/shared/kbn-data-source-catalog",
     "@kbn/data-sources-plugin": "link:x-pack/platform/plugins/shared/data_sources",
     "@kbn/data-stream-adapter": "link:x-pack/solutions/security/packages/data-stream-adapter",
     "@kbn/data-streams": "link:src/platform/packages/private/kbn-data-streams",

tsconfig.base.json

@@ -924,6 +924,8 @@
       "@kbn/data-service/*": ["src/platform/packages/shared/kbn-data-service/*"],
       "@kbn/data-service-server": ["src/platform/packages/shared/kbn-data-service-server"],
       "@kbn/data-service-server/*": ["src/platform/packages/shared/kbn-data-service-server/*"],
+      "@kbn/data-source-catalog": ["x-pack/platform/packages/shared/kbn-data-source-catalog"],
+      "@kbn/data-source-catalog/*": ["x-pack/platform/packages/shared/kbn-data-source-catalog/*"],
       "@kbn/data-sources-plugin": ["x-pack/platform/plugins/shared/data_sources"],
       "@kbn/data-sources-plugin/*": ["x-pack/platform/plugins/shared/data_sources/*"],
       "@kbn/data-stream-adapter": ["x-pack/solutions/security/packages/data-stream-adapter"],

x-pack/platform/packages/shared/kbn-data-source-catalog/README.md

@@ -0,0 +1,3 @@
+# @kbn/data-source-catalog
+
+Empty package generated by @kbn/generate

x-pack/platform/packages/shared/kbn-data-source-catalog/index.ts

@@ -0,0 +1,34 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+export type {
+  DataSourceEntry,
+  DataSourceType,
+  DataSourceStats,
+  DataSourceSemantic,
+  FreshnessCategory,
+  FieldMetadata,
+  IntegrationMetadata,
+  CatalogQueryParams,
+  CatalogQueryResult,
+} from './src/types';
+
+export {
+  CATALOG_INDEX_NAME,
+  DEFAULT_FIELD_LIMIT,
+  DEFAULT_SECURITY_PATTERNS,
+  FRESHNESS_THRESHOLDS,
+  CATALOG_VERSION,
+} from './src/constants';
+
+export { catalogIndexMapping } from './src/index_mapping';
+
+export { CatalogClient } from './src/catalog_client';
+export { CatalogQuery } from './src/catalog_query';
+export { refreshCatalog } from './src/catalog_refresh';
+export type { PackageClientLike } from './src/providers/integration_provider';
+export { generateHeuristicSummary } from './src/providers/heuristic_summary_provider';

x-pack/platform/packages/shared/kbn-data-source-catalog/jest.config.js

@@ -0,0 +1,12 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+module.exports = {
+  preset: '@kbn/test/jest_node',
+  rootDir: '../../../../..',
+  roots: ['<rootDir>/x-pack/platform/packages/shared/kbn-data-source-catalog'],
+};

x-pack/platform/packages/shared/kbn-data-source-catalog/kibana.jsonc

@@ -0,0 +1,9 @@
+{
+  "type": "shared-server",
+  "id": "@kbn/data-source-catalog",
+  "owner": [
+    "@elastic/security-detection-engine"
+  ],
+  "group": "platform",
+  "visibility": "shared"
+}

x-pack/platform/packages/shared/kbn-data-source-catalog/package.json

@@ -0,0 +1,7 @@
+{
+  "name": "@kbn/data-source-catalog",
+  "private": true,
+  "version": "1.0.0",
+  "license": "Elastic License 2.0 OR AGPL-3.0-only OR SSPL-1.0",
+  "sideEffects": false
+}

x-pack/platform/packages/shared/kbn-data-source-catalog/src/catalog_client.test.ts

@@ -0,0 +1,132 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { elasticsearchServiceMock } from '@kbn/core/server/mocks';
+import { CatalogClient } from './catalog_client';
+import { CATALOG_INDEX_NAME } from './constants';
+import { catalogIndexMapping } from './index_mapping';
+import type { DataSourceEntry } from './types';
+
+const makeEntry = (id: string): DataSourceEntry => ({
+  id,
+  name: `entry-${id}`,
+  type: 'index',
+  mapping: {
+    fields: [],
+    total_field_count: 0,
+    ecs_field_count: 0,
+    ecs_field_coverage: 0,
+  },
+  catalog_version: 1,
+  refreshed_at: '2024-01-01T00:00:00.000Z',
+});
+
+describe('CatalogClient', () => {
+  let esClient: ReturnType<typeof elasticsearchServiceMock.createElasticsearchClient>;
+  let catalogClient: CatalogClient;
+
+  beforeEach(() => {
+    esClient = elasticsearchServiceMock.createElasticsearchClient();
+    catalogClient = new CatalogClient(esClient);
+  });
+
+  describe('ensureIndex', () => {
+    it('creates the index if it does not exist', async () => {
+      esClient.indices.exists.mockResolvedValue(false);
+
+      await catalogClient.ensureIndex();
+
+      expect(esClient.indices.exists).toHaveBeenCalledWith({ index: CATALOG_INDEX_NAME });
+      expect(esClient.indices.create).toHaveBeenCalledWith({
+        index: CATALOG_INDEX_NAME,
+        mappings: catalogIndexMapping,
+        settings: {
+          number_of_shards: 1,
+          auto_expand_replicas: '0-1',
+        },
+      });
+    });
+
+    it('skips creation if index already exists', async () => {
+      esClient.indices.exists.mockResolvedValue(true);
+
+      await catalogClient.ensureIndex();
+
+      expect(esClient.indices.exists).toHaveBeenCalledWith({ index: CATALOG_INDEX_NAME });
+      expect(esClient.indices.create).not.toHaveBeenCalled();
+    });
+
+    it('does not throw when another node created the index concurrently', async () => {
+      esClient.indices.exists.mockResolvedValue(false);
+      const raceError = Object.assign(new Error('resource_already_exists_exception'), {
+        meta: { body: { error: { type: 'resource_already_exists_exception' } } },
+      });
+      esClient.indices.create.mockRejectedValue(raceError);
+
+      await expect(catalogClient.ensureIndex()).resolves.toBeUndefined();
+    });
+
+    it('re-throws non-race-condition errors from index creation', async () => {
+      esClient.indices.exists.mockResolvedValue(false);
+      const unexpectedError = new Error('unknown_error');
+      esClient.indices.create.mockRejectedValue(unexpectedError);
+
+      await expect(catalogClient.ensureIndex()).rejects.toThrow('unknown_error');
+    });
+  });
+
+  describe('bulkUpsert', () => {
+    it('indexes entries via bulk API with correct _id and _index', async () => {
+      esClient.bulk.mockResolvedValue({ errors: false, took: 1, items: [] });
+
+      const entries = [makeEntry('a'), makeEntry('b')];
+      await catalogClient.bulkUpsert(entries);
+
+      expect(esClient.bulk).toHaveBeenCalledWith({
+        operations: [
+          { index: { _index: CATALOG_INDEX_NAME, _id: 'a' } },
+          entries[0],
+          { index: { _index: CATALOG_INDEX_NAME, _id: 'b' } },
+          entries[1],
+        ],
+        refresh: 'wait_for',
+      });
+    });
+
+    it('throws when bulk response has errors', async () => {
+      esClient.bulk.mockResolvedValue({
+        errors: true,
+        took: 1,
+        items: [
+          { index: { _index: CATALOG_INDEX_NAME, _id: 'a', status: 400, error: { type: 'mapper_parsing_exception', reason: 'mapping error' } } },
+        ],
+      });
+
+      await expect(catalogClient.bulkUpsert([makeEntry('a')])).rejects.toThrow(
+        'Bulk upsert had errors: mapping error'
+      );
+    });
+
+    it('returns early for empty array without calling bulk', async () => {
+      await catalogClient.bulkUpsert([]);
+
+      expect(esClient.bulk).not.toHaveBeenCalled();
+    });
+  });
+
+  describe('deleteAll', () => {
+    it('calls deleteByQuery with match_all on the catalog index', async () => {
+      await catalogClient.deleteAll();
+
+      expect(esClient.deleteByQuery).toHaveBeenCalledWith({
+        index: CATALOG_INDEX_NAME,
+        query: { match_all: {} },
+        refresh: true,
+      });
+    });
+  });
+});

x-pack/platform/packages/shared/kbn-data-source-catalog/src/catalog_client.ts

@@ -0,0 +1,71 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import type { ElasticsearchClient } from '@kbn/core-elasticsearch-server';
+import { CATALOG_INDEX_NAME } from './constants';
+import { catalogIndexMapping } from './index_mapping';
+import type { DataSourceEntry } from './types';
+
+export class CatalogClient {
+  constructor(private readonly esClient: ElasticsearchClient) {}
+
+  async ensureIndex(): Promise<void> {
+    const exists = await this.esClient.indices.exists({ index: CATALOG_INDEX_NAME });
+    if (exists) {
+      return;
+    }
+    try {
+      await this.esClient.indices.create({
+        index: CATALOG_INDEX_NAME,
+        mappings: catalogIndexMapping,
+        settings: {
+          number_of_shards: 1,
+          auto_expand_replicas: '0-1',
+        },
+      });
+    } catch (error: unknown) {
+      // Handle race condition: another Kibana node may have created the index
+      const isAlreadyExists =
+        error instanceof Error &&
+        'meta' in error &&
+        (error as { meta?: { body?: { error?: { type?: string } } } }).meta?.body?.error?.type ===
+          'resource_already_exists_exception';
+      if (!isAlreadyExists) {
+        throw error;
+      }
+    }
+  }
+
+  async bulkUpsert(entries: DataSourceEntry[]): Promise<void> {
+    if (entries.length === 0) {
+      return;
+    }
+
+    const operations = entries.flatMap((entry) => [
+      { index: { _index: CATALOG_INDEX_NAME, _id: entry.id } },
+      entry,
+    ]);
+
+    const result = await this.esClient.bulk({ operations, refresh: 'wait_for' });
+
+    if (result.errors) {
+      const errorItems = result.items
+        .filter((item) => item.index?.error)
+        .map((item) => item.index?.error?.reason)
+        .slice(0, 5);
+      throw new Error(`Bulk upsert had errors: ${errorItems.join('; ')}`);
+    }
+  }
+
+  async deleteAll(): Promise<void> {
+    await this.esClient.deleteByQuery({
+      index: CATALOG_INDEX_NAME,
+      query: { match_all: {} },
+      refresh: true,
+    });
+  }
+}