diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS index c30117937e84a..771a4ab3e8d50 100644 --- a/.github/CODEOWNERS +++ b/.github/CODEOWNERS @@ -1330,6 +1330,7 @@ packages/kbn-monaco/src/esql @elastic/kibana-esql /x-pack/solutions/observability/plugins/infra/server/routes/log_analysis @elastic/obs-ux-logs-team /x-pack/solutions/observability/plugins/infra/server/services/rules @elastic/obs-ux-infra_services-team @elastic/obs-ux-logs-team /x-pack/test/common/utils/synthtrace @elastic/obs-ux-infra_services-team @elastic/obs-ux-logs-team # Assigned per https://github.com/elastic/kibana/blob/main/packages/kbn-apm-synthtrace/kibana.jsonc#L5 +/x-pack/test/common/utils/server_route_repository @elastic/obs-knowledge-team # Infra Monitoring tests /x-pack/test/common/services/infra_synthtrace_kibana_client.ts @elastic/obs-ux-infra_services-team diff --git a/src/platform/packages/shared/kbn-server-route-repository-client/src/is_request_aborted_error.ts b/src/platform/packages/shared/kbn-server-route-repository-client/src/is_request_aborted_error.ts index 3ab33ebac821c..2964e14288da5 100644 --- a/src/platform/packages/shared/kbn-server-route-repository-client/src/is_request_aborted_error.ts +++ b/src/platform/packages/shared/kbn-server-route-repository-client/src/is_request_aborted_error.ts @@ -9,6 +9,6 @@ import { get } from 'lodash'; -export function isRequestAbortedError(error: unknown): error is Error { +export function isRequestAbortedError(error: unknown): error is Error & { name: 'AbortError' } { return get(error, 'name') === 'AbortError'; } diff --git a/src/platform/packages/shared/kbn-server-route-repository-utils/src/typings.ts b/src/platform/packages/shared/kbn-server-route-repository-utils/src/typings.ts index 5db4a87b8b326..6cc176113a590 100644 --- a/src/platform/packages/shared/kbn-server-route-repository-utils/src/typings.ts +++ b/src/platform/packages/shared/kbn-server-route-repository-utils/src/typings.ts @@ -213,7 +213,7 @@ type DecodedRequestParamsOfType = : never; export type EndpointOf = - keyof TServerRouteRepository; + keyof TServerRouteRepository & string; export type ReturnOf< TServerRouteRepository extends ServerRouteRepository, diff --git a/x-pack/packages/kbn-streams-schema/src/helpers/type_guards.ts b/x-pack/packages/kbn-streams-schema/src/helpers/type_guards.ts index 557513fa74bb2..d8d3091e91bde 100644 --- a/x-pack/packages/kbn-streams-schema/src/helpers/type_guards.ts +++ b/x-pack/packages/kbn-streams-schema/src/helpers/type_guards.ts @@ -63,15 +63,11 @@ export function isStream(subject: any): subject is StreamDefinition { return isSchema(streamDefintionSchema, subject); } -export function isIngestStream( - subject: IngestStreamDefinition | WiredStreamDefinition -): subject is IngestStreamDefinition { +export function isIngestStream(subject: StreamDefinition): subject is IngestStreamDefinition { return isSchema(ingestStreamDefinitonSchema, subject); } -export function isWiredStream( - subject: IngestStreamDefinition | WiredStreamDefinition -): subject is WiredStreamDefinition { +export function isWiredStream(subject: StreamDefinition): subject is WiredStreamDefinition { return isSchema(wiredStreamDefinitonSchema, subject); } diff --git a/x-pack/packages/kbn-streams-schema/src/models/streams/ingest_stream.ts b/x-pack/packages/kbn-streams-schema/src/models/streams/ingest_stream.ts index d21f11d869929..6c283c38c201f 100644 --- a/x-pack/packages/kbn-streams-schema/src/models/streams/ingest_stream.ts +++ b/x-pack/packages/kbn-streams-schema/src/models/streams/ingest_stream.ts @@ -14,6 +14,7 @@ export const ingestStreamDefinitonSchema = z name: z.string(), elasticsearch_assets: z.optional(elasticsearchAssetSchema), stream: ingestStreamConfigDefinitonSchema, + dashboards: z.optional(z.array(z.string())), }) .strict(); diff --git a/x-pack/packages/kbn-streams-schema/src/models/streams/wired_stream.ts b/x-pack/packages/kbn-streams-schema/src/models/streams/wired_stream.ts index 0374472673cdb..8a75a1a1c6a4f 100644 --- a/x-pack/packages/kbn-streams-schema/src/models/streams/wired_stream.ts +++ b/x-pack/packages/kbn-streams-schema/src/models/streams/wired_stream.ts @@ -14,6 +14,7 @@ export const wiredStreamDefinitonSchema = z name: z.string(), elasticsearch_assets: z.optional(elasticsearchAssetSchema), stream: wiredStreamConfigDefinitonSchema, + dashboards: z.optional(z.array(z.string())), }) .strict(); diff --git a/x-pack/platform/plugins/shared/alerting/server/application/rule/methods/find/schemas/find_rules_schemas.ts b/x-pack/platform/plugins/shared/alerting/server/application/rule/methods/find/schemas/find_rules_schemas.ts index aec95d7f2c061..4ea4afa8b0c79 100644 --- a/x-pack/platform/plugins/shared/alerting/server/application/rule/methods/find/schemas/find_rules_schemas.ts +++ b/x-pack/platform/plugins/shared/alerting/server/application/rule/methods/find/schemas/find_rules_schemas.ts @@ -7,6 +7,11 @@ import { schema } from '@kbn/config-schema'; +const savedObjectReferenceSchema = schema.object({ + type: schema.string(), + id: schema.string(), +}); + export const findRulesOptionsSchema = schema.object( { perPage: schema.maybe(schema.number()), @@ -19,10 +24,7 @@ export const findRulesOptionsSchema = schema.object( sortField: schema.maybe(schema.string()), sortOrder: schema.maybe(schema.oneOf([schema.literal('asc'), schema.literal('desc')])), hasReference: schema.maybe( - schema.object({ - type: schema.string(), - id: schema.string(), - }) + schema.oneOf([savedObjectReferenceSchema, schema.arrayOf(savedObjectReferenceSchema)]) ), fields: schema.maybe(schema.arrayOf(schema.string())), filter: schema.maybe( diff --git a/x-pack/solutions/observability/packages/utils_server/es/client/create_observability_es_client.ts b/x-pack/solutions/observability/packages/utils_server/es/client/create_observability_es_client.ts index 92c7b8d19e531..7731d72ffd0fe 100644 --- a/x-pack/solutions/observability/packages/utils_server/es/client/create_observability_es_client.ts +++ b/x-pack/solutions/observability/packages/utils_server/es/client/create_observability_es_client.ts @@ -24,7 +24,7 @@ import { esqlResultToPlainObjects } from '../esql_result_to_plain_objects'; type SearchRequest = ESSearchRequest & { index: string | string[]; track_total_hits: number | boolean; - size: number | boolean; + size: number; }; export interface EsqlOptions { @@ -112,10 +112,12 @@ export function createObservabilityEsClient({ client, logger, plugin, + labels, }: { client: ElasticsearchClient; logger: Logger; - plugin: string; + plugin?: string; + labels?: Record; }): ObservabilityElasticsearchClient { // wraps the ES calls in a named APM span for better analysis // (otherwise it would just eg be a _search span) @@ -129,7 +131,8 @@ export function createObservabilityEsClient({ { name: operationName, labels: { - plugin, + ...labels, + ...(plugin ? { plugin } : {}), }, }, callback, diff --git a/x-pack/solutions/observability/packages/utils_server/es/storage/README.md b/x-pack/solutions/observability/packages/utils_server/es/storage/README.md new file mode 100644 index 0000000000000..45f9015cad9fb --- /dev/null +++ b/x-pack/solutions/observability/packages/utils_server/es/storage/README.md @@ -0,0 +1,76 @@ +# Storage adapter + +Storage adapters are an abstraction for managing & writing data into Elasticsearch, from Kibana plugins. + +There are several ways one can use Elasticsearch in Kibana, for instance: + +- a simple id-based CRUD table +- timeseries data with regular indices +- timeseries data with data streams + +But then there are many choices to be made that make this a very complex problem: + +- Elasticsearch asset managmeent +- Authentication +- Schema changes +- Kibana's distributed nature +- Stateful versus serverless + +The intent of storage adapters is to come up with an abstraction that allows Kibana developers to have a common interface for writing to and reading data from Elasticsearch. For instance, for setting up your data store, it should not matter how you authenticate (internal user? current user? API keys?). + +## Saved objects + +Some of these problems are solved by Saved Objects. But Saved Objects come with a lot of baggage - Kibana RBAC, relationships, spaces, all of which might not be +needed for your use case but are still restrictive. One could consider Saved Objects to be the target of an adapter, but Storage Adapters aim to address a wider set of use-cases. + +## Philosophy + +Storage adapters should largely adhere to the following principles: + +- Interfaces are as close to Elasticsearch as possible. Meaning, the `search` method is practically a pass-through for `_search`. +- Strongly-typed. TypeScript types are inferred from the schema. This makes it easy to create fully-typed clients for any storage. +- Lazy writes. No Elasticsearch assets (templates, indices, aliases) get installed unless necessary. Anything that gets persisted to Elasticsearch raises questions (in SDHs, UIs, APIs) and should be avoided when possible. This also helps avoidable upgrade issues (e.g. conflicting mappings for something that never got used). +- Recoverable. If somehow Elasticsearch assets get borked, the adapters should make a best-effort attempt to recover, or log warnings with clear remediation steps. + +## Future goals + +Currently, we only have the StorageIndexAdapter which writes to plain indices. In the future, we'll want more: + +- A StorageDataStreamAdapter or StorageSavedObjectAdapter +- Federated search +- Data/Index Lifecycle Management +- Migration scripts +- Runtime mappings for older versions + +## Usage + +### Storage index adapter + +To use the storage index adapter, instantiate it with an authenticated Elasticsearch client: + +```ts + const storageSettings = { + name: '.kibana_streams_assets', + schema: { + properties: { + [ASSET_ASSET_ID]: types.keyword({ required: true }), + [ASSET_TYPE]: types.enum(Object.values(ASSET_TYPES), { required: true }), + }, + }, + } satisfies IndexStorageSettings; + + // create and configure the adapter + const adapter = new StorageIndexAdapter( + esClient: coreStart.elasticsearch.client.asInternalUser, + this.logger.get('assets'), + storageSettings + ); + + // get the client (its interface is shared across all adapters) + const client = adapter.getClient(); + + const response = await client.search('operation_name', { + track_total_hits: true + }); + +``` diff --git a/x-pack/solutions/observability/packages/utils_server/es/storage/get_schema_version.ts b/x-pack/solutions/observability/packages/utils_server/es/storage/get_schema_version.ts new file mode 100644 index 0000000000000..0be986c168cba --- /dev/null +++ b/x-pack/solutions/observability/packages/utils_server/es/storage/get_schema_version.ts @@ -0,0 +1,15 @@ +/* + * 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 stringify from 'json-stable-stringify'; +import objectHash from 'object-hash'; +import { IndexStorageSettings } from '.'; + +export function getSchemaVersion(storage: IndexStorageSettings): string { + const version = objectHash(stringify(storage.schema.properties)); + return version; +} diff --git a/x-pack/solutions/observability/packages/utils_server/es/storage/index.ts b/x-pack/solutions/observability/packages/utils_server/es/storage/index.ts new file mode 100644 index 0000000000000..913f079e93313 --- /dev/null +++ b/x-pack/solutions/observability/packages/utils_server/es/storage/index.ts @@ -0,0 +1,91 @@ +/* + * 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 { + BulkOperationContainer, + BulkRequest, + BulkResponse, + DeleteRequest, + DeleteResponse, + IndexRequest, + IndexResponse, + SearchRequest, +} from '@elastic/elasticsearch/lib/api/types'; +import { InferSearchResponseOf } from '@kbn/es-types'; +import { StorageFieldTypeOf, StorageMappingProperty } from './types'; + +interface StorageSchemaProperties { + [x: string]: StorageMappingProperty; +} + +export interface StorageSchema { + properties: StorageSchemaProperties; +} + +interface StorageSettingsBase { + schema: StorageSchema; +} + +export interface IndexStorageSettings extends StorageSettingsBase { + name: string; +} + +export type StorageSettings = IndexStorageSettings; + +export type StorageAdapterSearchRequest = Omit; +export type StorageAdapterSearchResponse< + TDocument, + TSearchRequest extends Omit +> = InferSearchResponseOf; + +export type StorageAdapterBulkOperation = Pick; + +export type StorageAdapterBulkRequest> = Omit< + BulkRequest, + 'operations' | 'index' +> & { + operations: Array; +}; +export type StorageAdapterBulkResponse = BulkResponse; + +export type StorageAdapterDeleteRequest = DeleteRequest; +export type StorageAdapterDeleteResponse = DeleteResponse; + +export type StorageAdapterIndexRequest = Omit< + IndexRequest, + 'index' +>; +export type StorageAdapterIndexResponse = IndexResponse; + +export interface IStorageAdapter { + bulk>( + request: StorageAdapterBulkRequest + ): Promise; + search>( + request: StorageAdapterSearchRequest + ): Promise>; + index( + request: StorageAdapterIndexRequest + ): Promise; + delete(request: StorageAdapterDeleteRequest): Promise; +} + +export type StorageSettingsOf> = + TStorageAdapter extends IStorageAdapter + ? TStorageSettings extends StorageSettings + ? TStorageSettings + : never + : never; + +export type StorageDocumentOf = { + [TKey in keyof TStorageSettings['schema']['properties']]: StorageFieldTypeOf< + TStorageSettings['schema']['properties'][TKey] + >; +} & { _id: string }; + +export { StorageIndexAdapter } from './index_adapter'; +export { StorageClient } from './storage_client'; +export { types } from './types'; diff --git a/x-pack/solutions/observability/packages/utils_server/es/storage/index_adapter/index.test.ts b/x-pack/solutions/observability/packages/utils_server/es/storage/index_adapter/index.test.ts new file mode 100644 index 0000000000000..061ccfdaac219 --- /dev/null +++ b/x-pack/solutions/observability/packages/utils_server/es/storage/index_adapter/index.test.ts @@ -0,0 +1,588 @@ +/* + * 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 { errors } from '@elastic/elasticsearch'; +import { + BulkDeleteOperation, + BulkIndexOperation, + BulkRequest, + BulkResponseItem, + IndexRequest, + IndicesGetAliasIndexAliases, + IndicesIndexState, + IndicesPutIndexTemplateRequest, + IndicesSimulateIndexTemplateResponse, + SearchRequest, + SearchResponse, +} from '@elastic/elasticsearch/lib/api/types'; +import { ElasticsearchClient, Logger } from '@kbn/core/server'; +import { castArray, merge, remove } from 'lodash'; +import { Required } from 'utility-types'; +import { v4 } from 'uuid'; +import { StorageIndexAdapter } from '.'; +import { StorageSettings } from '..'; +import * as getSchemaVersionModule from '../get_schema_version'; + +type MockedElasticsearchClient = jest.Mocked & { + indices: jest.Mocked; +}; + +const createEsClientMock = (): MockedElasticsearchClient => { + return { + indices: { + putIndexTemplate: jest.fn(), + getIndexTemplate: jest.fn(), + create: jest.fn(), + getAlias: jest.fn(), + putAlias: jest.fn(), + existsIndexTemplate: jest.fn(), + existsAlias: jest.fn(), + exists: jest.fn(), + get: jest.fn(), + simulateIndexTemplate: jest.fn(), + putMapping: jest.fn(), + putSettings: jest.fn(), + }, + search: jest.fn(), + bulk: jest.fn(), + index: jest.fn(), + delete: jest.fn(), + } as unknown as MockedElasticsearchClient; +}; + +const createLoggerMock = (): jest.Mocked => { + const logger = { + debug: jest.fn(), + info: jest.fn(), + error: jest.fn(), + get: jest.fn(), + } as unknown as jest.Mocked; + + logger.get.mockReturnValue(logger); + + return logger; +}; + +const TEST_INDEX_NAME = 'test_index'; + +function getIndexName(counter: number) { + return `${TEST_INDEX_NAME}-00000${counter.toString()}`; +} + +describe('StorageIndexAdapter', () => { + let esClientMock: MockedElasticsearchClient; + let loggerMock: jest.Mocked; + let adapter: StorageIndexAdapter; + + const storageSettings = { + name: TEST_INDEX_NAME, + schema: { + properties: { + foo: { + type: 'keyword', + required: true, + }, + }, + }, + } satisfies StorageSettings; + + beforeEach(() => { + esClientMock = createEsClientMock(); + loggerMock = createLoggerMock(); + + adapter = new StorageIndexAdapter(esClientMock, loggerMock, storageSettings); + + mockEmptyState(); + + mockCreateAPIs(); + + jest.spyOn(getSchemaVersionModule, 'getSchemaVersion').mockReturnValue('current_version'); + }); + + afterEach(() => { + jest.clearAllMocks(); + }); + + it('creates a named logger', () => { + expect(loggerMock.get).toHaveBeenCalledWith('storage'); + expect(loggerMock.get).toHaveBeenCalledWith('test_index'); + }); + + it('does not install index templates or backing indices initially', () => { + expect(esClientMock.indices.putIndexTemplate).not.toHaveBeenCalled(); + expect(esClientMock.indices.create).not.toHaveBeenCalled(); + }); + + it('does not install index templates or backing indices after searching', async () => { + const mockSearchResponse = { + hits: { + hits: [{ _id: 'doc1', _source: { foo: 'bar' } }], + }, + } as unknown as SearchResponse<{ foo: 'bar' }>; + + esClientMock.search.mockResolvedValueOnce(mockSearchResponse); + + await adapter.search({ query: { match_all: {} } }); + + expect(esClientMock.indices.putIndexTemplate).not.toHaveBeenCalled(); + expect(esClientMock.indices.create).not.toHaveBeenCalled(); + }); + + it('does not fail a search when an index does not exist', async () => { + const mockSearchResponse = { + hits: { + hits: [], + }, + } as unknown as SearchResponse; + + esClientMock.search.mockResolvedValueOnce(mockSearchResponse); + + await adapter.search({ query: { match_all: {} } }); + + expect(esClientMock.search).toHaveBeenCalledWith( + expect.objectContaining({ + allow_no_indices: true, + }) + ); + }); + + describe('when writing/bootstrapping without an existing index', () => { + function verifyResources() { + expect(esClientMock.indices.putIndexTemplate).toHaveBeenCalled(); + expect(esClientMock.indices.create).toHaveBeenCalledWith( + expect.objectContaining({ + index: 'test_index-000001', + }) + ); + } + + describe('when using index', () => { + it('creates the resources', async () => { + await adapter.index({ id: 'doc1', document: { foo: 'bar' } }); + + verifyResources(); + + expect(esClientMock.index).toHaveBeenCalledTimes(1); + }); + }); + + describe('when using bulk', () => { + it('creates the resources', async () => { + await adapter.bulk({ + operations: [{ index: { _id: 'foo' } }, { foo: 'bar' }], + }); + + verifyResources(); + + expect(esClientMock.bulk).toHaveBeenCalledTimes(1); + }); + }); + }); + + describe('when writing/bootstrapping with an existing, compatible index', () => { + beforeEach(async () => { + await esClientMock.indices.putIndexTemplate({ + name: TEST_INDEX_NAME, + _meta: { + version: 'current_version', + }, + template: { + mappings: { + _meta: { + version: 'current_version', + }, + properties: { + foo: { type: 'keyword', meta: { required: 'true' } }, + }, + }, + }, + }); + + await esClientMock.indices.create({ + index: getIndexName(1), + }); + + esClientMock.indices.putIndexTemplate.mockClear(); + esClientMock.indices.create.mockClear(); + }); + + it('does not recreate or update index template', async () => { + await adapter.index({ id: 'doc2', document: { foo: 'bar' } }); + + expect(esClientMock.indices.putIndexTemplate).not.toHaveBeenCalled(); + expect(esClientMock.indices.create).not.toHaveBeenCalled(); + + expect(esClientMock.index).toHaveBeenCalledWith( + expect.objectContaining({ + index: 'test_index', + id: 'doc2', + }) + ); + }); + }); + + describe('when writing/bootstrapping with an existing, outdated index', () => { + beforeEach(async () => { + await esClientMock.indices.putIndexTemplate({ + name: TEST_INDEX_NAME, + _meta: { + version: 'first_version', + }, + template: { + mappings: { + _meta: { + version: 'first_version', + }, + properties: {}, + }, + }, + }); + + await esClientMock.indices.create({ + index: getIndexName(1), + }); + + esClientMock.indices.putIndexTemplate.mockClear(); + esClientMock.indices.create.mockClear(); + esClientMock.indices.simulateIndexTemplate.mockClear(); + }); + + it('updates index mappings on write', async () => { + await adapter.index({ id: 'docY', document: { foo: 'bar' } }); + + expect(esClientMock.indices.putIndexTemplate).toHaveBeenCalled(); + expect(esClientMock.indices.simulateIndexTemplate).toHaveBeenCalled(); + + expect(esClientMock.indices.putMapping).toHaveBeenCalledWith( + expect.objectContaining({ + properties: { + foo: { + type: 'keyword', + meta: { + multi_value: 'false', + required: 'true', + }, + }, + }, + }) + ); + }); + }); + + describe('when indexing', () => { + describe('a new document', () => { + it('indexes the document via alias with require_alias=true', async () => { + const res = await adapter.index({ id: 'doc_1', document: { foo: 'bar' } }); + + expect(esClientMock.index).toHaveBeenCalledWith( + expect.objectContaining({ + index: 'test_index', + require_alias: true, + id: 'doc_1', + document: { foo: 'bar' }, + }) + ); + + expect(res._index).toBe('test_index-000001'); + expect(res._id).toBe('doc_1'); + }); + }); + + describe('an existing document in any non-write index', () => { + beforeEach(async () => { + await adapter.index({ id: 'doc_1', document: { foo: 'bar' } }); + + await esClientMock.indices.create({ + index: getIndexName(2), + }); + }); + + it('deletes the dangling item from non-write indices', async () => { + await adapter.index({ id: 'doc_1', document: { foo: 'bar' } }); + + expect(esClientMock.delete).toHaveBeenCalledWith( + expect.objectContaining({ + index: 'test_index-000001', + id: 'doc_1', + }) + ); + + expect(esClientMock.index).toHaveBeenCalledWith( + expect.objectContaining({ + index: 'test_index', + id: 'doc_1', + }) + ); + }); + }); + }); + + describe('when bulk indexing', () => { + describe('an existing document in any non-write index', () => { + beforeEach(async () => { + await adapter.bulk({ + operations: [ + { + index: { + _id: 'doc_1', + }, + }, + { + foo: 'bar', + }, + ], + }); + + await esClientMock.indices.create({ + index: getIndexName(2), + }); + }); + + it('deletes the dangling item from non-write indices', async () => { + await adapter.bulk({ + operations: [ + { + index: { + _id: 'doc_1', + }, + }, + { + foo: 'bar', + }, + ], + }); + + expect(esClientMock.bulk).toHaveBeenLastCalledWith( + expect.objectContaining({ + index: 'test_index', + operations: [ + { + index: { + _id: 'doc_1', + }, + }, + { + foo: 'bar', + }, + { + delete: { + _index: getIndexName(1), + _id: 'doc_1', + }, + }, + ], + }) + ); + }); + }); + }); + + function mockEmptyState() { + esClientMock.indices.getIndexTemplate.mockResolvedValue({ + index_templates: [], + }); + esClientMock.indices.existsIndexTemplate.mockResolvedValue(false); + + esClientMock.indices.simulateIndexTemplate.mockImplementation(async () => { + throw new errors.ResponseError({ statusCode: 404, warnings: [], meta: {} as any }); + }); + + esClientMock.indices.existsAlias.mockResolvedValue(false); + esClientMock.indices.exists.mockResolvedValue(false); + esClientMock.indices.getAlias.mockResolvedValue({}); + + esClientMock.index.mockImplementation(async () => { + throw new errors.ResponseError({ statusCode: 404, warnings: [], meta: {} as any }); + }); + + esClientMock.bulk.mockImplementation(async () => { + throw new errors.ResponseError({ statusCode: 404, warnings: [], meta: {} as any }); + }); + } + + function mockCreateAPIs() { + const indices: Map< + string, + Pick + > = new Map(); + + const docs: Array<{ _id: string; _source: Record; _index: string }> = []; + + function getCurrentWriteIndex() { + return Array.from(indices.entries()).find( + ([indexName, indexState]) => indexState.aliases![TEST_INDEX_NAME].is_write_index + )?.[0]; + } + + esClientMock.indices.putIndexTemplate.mockImplementation(async (_templateRequest) => { + const templateRequest = _templateRequest as Required< + IndicesPutIndexTemplateRequest, + 'template' + >; + + esClientMock.indices.existsIndexTemplate.mockResolvedValue(true); + esClientMock.indices.getIndexTemplate.mockResolvedValue({ + index_templates: [ + { + name: templateRequest.name, + index_template: { + _meta: templateRequest._meta, + template: templateRequest.template, + index_patterns: `${TEST_INDEX_NAME}*`, + composed_of: [], + }, + }, + ], + }); + + esClientMock.indices.simulateIndexTemplate.mockImplementation( + async (mockSimulateRequest): Promise => { + return { + template: { + aliases: templateRequest.template?.aliases ?? {}, + mappings: templateRequest.template?.mappings ?? {}, + settings: templateRequest.template?.settings ?? {}, + }, + }; + } + ); + + esClientMock.indices.create.mockImplementation(async (createIndexRequest) => { + const indexName = createIndexRequest.index; + + const prevIndices = Array.from(indices.entries()); + + prevIndices.forEach(([currentIndexName, indexState]) => { + indexState.aliases![TEST_INDEX_NAME] = { + is_write_index: false, + }; + }); + + indices.set(indexName, { + aliases: merge({}, templateRequest.template.aliases ?? {}, { + [TEST_INDEX_NAME]: { is_write_index: true }, + }), + mappings: templateRequest.template.mappings ?? {}, + settings: templateRequest.template.settings ?? {}, + }); + + esClientMock.indices.getAlias.mockImplementation(async (aliasRequest) => { + return Object.fromEntries( + Array.from(indices.entries()).map(([currentIndexName, indexState]) => { + return [ + currentIndexName, + { aliases: indexState.aliases ?? {} } satisfies IndicesGetAliasIndexAliases, + ]; + }) + ); + }); + + esClientMock.indices.get.mockImplementation(async () => { + return Object.fromEntries(indices.entries()); + }); + + esClientMock.index.mockImplementation(async (_indexRequest) => { + const indexRequest = _indexRequest as IndexRequest; + const id = indexRequest.id ?? v4(); + const index = getCurrentWriteIndex()!; + + docs.push({ + _id: id, + _index: index, + _source: indexRequest.document!, + }); + + return { + _id: id, + _index: index, + _shards: { + failed: 0, + successful: 1, + total: 1, + }, + _version: 1, + result: 'created', + }; + }); + + esClientMock.search.mockImplementation(async (_searchRequest) => { + const searchRequest = _searchRequest as SearchRequest; + + const ids = castArray(searchRequest.query?.bool?.filter ?? [])?.[0]?.terms + ?._id as string[]; + + const excludeIndex = castArray(searchRequest.query?.bool?.must_not)?.[0]?.term + ?._index as string; + + const matches = docs.filter((doc) => { + return ids.includes(doc._id) && doc._index !== excludeIndex; + }); + + return { + took: 0, + timed_out: false, + _shards: { + successful: 1, + failed: 0, + total: 1, + }, + hits: { + hits: matches, + total: { + value: matches.length, + relation: 'eq', + }, + }, + }; + }); + + esClientMock.bulk.mockImplementation(async (_bulkRequest) => { + const bulkRequest = _bulkRequest as BulkRequest>; + + const items: Array>> = []; + + bulkRequest.operations?.forEach((operation, index, operations) => { + if ('index' in operation) { + const indexOperation = operation.index as BulkIndexOperation; + const document = { + _id: indexOperation._id ?? v4(), + _index: indexOperation._index ?? getCurrentWriteIndex()!, + _source: operations[index + 1], + }; + docs.push(document); + + items.push({ index: { _id: document._id, _index: document._index, status: 200 } }); + } else if ('delete' in operation) { + const deleteOperation = operation.delete as BulkDeleteOperation; + remove(docs, (doc) => { + return doc._id === deleteOperation._id && doc._index === deleteOperation._index; + }); + + items.push({ + delete: { + _id: deleteOperation._id!, + _index: deleteOperation._index!, + status: 200, + }, + }); + } + }); + + return { + errors: false, + took: 0, + items, + }; + }); + + return { acknowledged: true, index: createIndexRequest.index, shards_acknowledged: true }; + }); + + return { acknowledged: true }; + }); + } +}); diff --git a/x-pack/solutions/observability/packages/utils_server/es/storage/index_adapter/index.ts b/x-pack/solutions/observability/packages/utils_server/es/storage/index_adapter/index.ts new file mode 100644 index 0000000000000..258df256ed804 --- /dev/null +++ b/x-pack/solutions/observability/packages/utils_server/es/storage/index_adapter/index.ts @@ -0,0 +1,388 @@ +/* + * 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 { + IndexResponse, + IndicesIndexState, + IndicesIndexTemplate, + IndicesPutIndexTemplateIndexTemplateMapping, + MappingProperty, + SearchRequest, +} from '@elastic/elasticsearch/lib/api/types'; +import { ElasticsearchClient, Logger } from '@kbn/core/server'; +import { isResponseError } from '@kbn/es-errors'; +import { InferSearchResponseOf } from '@kbn/es-types'; +import { last, mapValues, padStart } from 'lodash'; +import { + IStorageAdapter, + IndexStorageSettings, + StorageAdapterBulkRequest, + StorageAdapterBulkResponse, + StorageAdapterDeleteRequest, + StorageAdapterDeleteResponse, + StorageAdapterIndexRequest, + StorageAdapterIndexResponse, + StorageAdapterSearchRequest, + StorageAdapterSearchResponse, +} from '..'; +import { getSchemaVersion } from '../get_schema_version'; +import { StorageClient } from '../storage_client'; +import { StorageMappingProperty } from '../types'; + +function getAliasName(name: string) { + return name; +} + +function getBackingIndexPattern(name: string) { + return `${name}-*`; +} + +function getBackingIndexName(name: string, count: number) { + const countId = padStart(count.toString(), 6, '0'); + return `${name}-${countId}`; +} + +function getIndexTemplateName(name: string) { + return `${name}`; +} + +function toElasticsearchMappingProperty(property: StorageMappingProperty): MappingProperty { + const { required, multi_value: multiValue, enum: enums, ...rest } = property; + + return { + ...rest, + meta: { + ...property.meta, + required: JSON.stringify(required ?? false), + multi_value: JSON.stringify(multiValue ?? false), + ...(enums ? { enum: JSON.stringify(enums) } : {}), + }, + }; +} + +function catchConflictError(error: Error) { + if (isResponseError(error) && error.statusCode === 409) { + return; + } + throw error; +} + +/** + * Adapter for writing and reading documents to/from Elasticsearch, + * using plain indices. + * + * TODO: + * - Index Lifecycle Management + * - Schema upgrades w/ fallbacks + */ +export class StorageIndexAdapter + implements IStorageAdapter +{ + private readonly logger: Logger; + constructor( + private readonly esClient: ElasticsearchClient, + logger: Logger, + private readonly storage: TStorageSettings + ) { + this.logger = logger.get('storage').get(this.storage.name); + } + + private getSearchIndexPattern(): string { + return `${getAliasName(this.storage.name)}*`; + } + + private getWriteTarget(): string { + return getAliasName(this.storage.name); + } + + private async createOrUpdateIndexTemplate(): Promise { + const version = getSchemaVersion(this.storage); + + const template: IndicesPutIndexTemplateIndexTemplateMapping = { + mappings: { + _meta: { + version, + }, + properties: mapValues(this.storage.schema.properties, toElasticsearchMappingProperty), + }, + aliases: { + [getAliasName(this.storage.name)]: { + is_write_index: true, + }, + }, + }; + + await this.esClient.indices + .putIndexTemplate({ + name: getIndexTemplateName(this.storage.name), + create: false, + allow_auto_create: false, + index_patterns: getBackingIndexPattern(this.storage.name), + _meta: { + version, + }, + template, + }) + .catch(catchConflictError); + } + + private async getExistingIndexTemplate(): Promise { + return await this.esClient.indices + .getIndexTemplate({ + name: getIndexTemplateName(this.storage.name), + }) + .then((templates) => templates.index_templates[0]?.index_template) + .catch((error) => { + if (isResponseError(error) && error.statusCode === 404) { + return undefined; + } + throw error; + }); + } + + private async getCurrentWriteIndex(): Promise< + { name: string; state: IndicesIndexState } | undefined + > { + const [writeIndex, indices] = await Promise.all([ + this.getCurrentWriteIndexName(), + this.getExistingIndices(), + ]); + + return writeIndex ? { name: writeIndex, state: indices[writeIndex] } : undefined; + } + + private async getExistingIndices() { + return this.esClient.indices.get({ + index: getBackingIndexPattern(this.storage.name), + allow_no_indices: true, + }); + } + + private async getCurrentWriteIndexName(): Promise { + const aliasName = getAliasName(this.storage.name); + + const aliases = await this.esClient.indices + .getAlias({ + name: getAliasName(this.storage.name), + }) + .catch((error) => { + if (isResponseError(error) && error.statusCode === 404) { + return {}; + } + throw error; + }); + + const writeIndex = Object.entries(aliases) + .map(([name, alias]) => { + return { + name, + isWriteIndex: alias.aliases[aliasName]?.is_write_index === true, + }; + }) + .find(({ isWriteIndex }) => { + return isWriteIndex; + }); + + return writeIndex?.name; + } + + private async createNextBackingIndex(): Promise { + const writeIndex = await this.getCurrentWriteIndexName(); + + const nextIndexName = getBackingIndexName( + this.storage.name, + writeIndex ? parseInt(last(writeIndex.split('-'))!, 10) : 1 + ); + + await this.esClient.indices + .create({ + index: nextIndexName, + }) + .catch(catchConflictError); + } + + private async updateMappingsOfExistingIndex({ name }: { name: string }) { + const simulateIndexTemplateResponse = await this.esClient.indices.simulateIndexTemplate({ + name: getIndexTemplateName(this.storage.name), + }); + + if (simulateIndexTemplateResponse.template.settings) { + await this.esClient.indices.putSettings({ + index: name, + settings: simulateIndexTemplateResponse.template.settings, + }); + } + + if (simulateIndexTemplateResponse.template.mappings) { + await this.esClient.indices.putMapping({ + index: name, + ...simulateIndexTemplateResponse.template.mappings, + }); + } + } + + /** + * Validates whether: + * - an index template exists + * - the index template has the right version (if not, update it) + * - a write index exists (if it doesn't, create it) + * - the write index has the right version (if not, update it) + */ + private async validateComponentsBeforeWriting(cb: () => Promise): Promise { + const [writeIndex, existingIndexTemplate] = await Promise.all([ + this.getCurrentWriteIndex(), + this.getExistingIndexTemplate(), + ]); + + const expectedSchemaVersion = getSchemaVersion(this.storage); + + if (!existingIndexTemplate) { + this.logger.info(`Creating index template as it does not exist`); + await this.createOrUpdateIndexTemplate(); + } else if (existingIndexTemplate._meta?.version !== expectedSchemaVersion) { + this.logger.info(`Updating existing index template`); + await this.createOrUpdateIndexTemplate(); + } + + if (!writeIndex) { + this.logger.info(`Creating first backing index`); + await this.createNextBackingIndex(); + } else if (writeIndex?.state.mappings?._meta?.version !== expectedSchemaVersion) { + this.logger.info(`Updating mappings of existing write index due to schema version mismatch`); + await this.updateMappingsOfExistingIndex({ + name: writeIndex.name, + }); + } + + return await cb(); + } + + async search>( + request: StorageAdapterSearchRequest + ): Promise> { + return this.esClient.search({ + ...request, + index: this.getSearchIndexPattern(), + allow_no_indices: true, + }) as unknown as Promise>; + } + + /** + * Get items from all non-write indices for the specified ids. + */ + private async getDanglingItems({ ids }: { ids: string[] }) { + const writeIndex = await this.getCurrentWriteIndexName(); + + if (writeIndex && ids.length) { + const danglingItemsResponse = await this.search({ + query: { + bool: { + filter: [{ terms: { _id: ids } }], + must_not: [ + { + term: { + _index: writeIndex, + }, + }, + ], + }, + }, + size: 10_000, + }); + + return danglingItemsResponse.hits.hits.map((hit) => ({ + id: hit._id!, + index: hit._index, + })); + } + return []; + } + + async index(request: StorageAdapterIndexRequest): Promise { + const attemptIndex = async (): Promise => { + const [danglingItem] = request.id + ? await this.getDanglingItems({ ids: [request.id] }) + : [undefined]; + + if (danglingItem) { + await this.esClient.delete({ + id: danglingItem.id, + index: danglingItem.index, + refresh: false, + }); + } + + return this.esClient.index({ + ...request, + refresh: request.refresh, + index: this.getWriteTarget(), + require_alias: true, + }); + }; + + return this.validateComponentsBeforeWriting(attemptIndex).then(async (response) => { + this.logger.debug(() => `Indexed document ${request.id} into ${response._index}`); + + return response; + }); + } + + async bulk>( + request: StorageAdapterBulkRequest + ): Promise { + const attemptBulk = async () => { + const indexedIds = + request.operations?.flatMap((operation) => { + if ( + 'index' in operation && + operation.index && + typeof operation.index === 'object' && + '_id' in operation.index && + typeof operation.index._id === 'string' + ) { + return operation.index._id ?? []; + } + return []; + }) ?? []; + + const danglingItems = await this.getDanglingItems({ ids: indexedIds }); + + if (danglingItems.length) { + this.logger.debug(`Deleting ${danglingItems.length} dangling items`); + } + + return this.esClient.bulk({ + ...request, + operations: (request.operations || []).concat( + danglingItems.map((item) => ({ delete: { _index: item.index, _id: item.id } })) + ), + index: this.getWriteTarget(), + require_alias: true, + }); + }; + + return this.validateComponentsBeforeWriting(attemptBulk).then(async (response) => { + return response; + }); + } + + async delete({ + id, + index, + refresh, + }: StorageAdapterDeleteRequest): Promise { + return await this.esClient.delete({ + index, + id, + refresh, + }); + } + + getClient(): StorageClient { + return new StorageClient(this, this.logger); + } +} diff --git a/x-pack/solutions/observability/packages/utils_server/es/storage/storage_client.ts b/x-pack/solutions/observability/packages/utils_server/es/storage/storage_client.ts new file mode 100644 index 0000000000000..6bd2b211d1083 --- /dev/null +++ b/x-pack/solutions/observability/packages/utils_server/es/storage/storage_client.ts @@ -0,0 +1,115 @@ +/* + * 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 { withSpan } from '@kbn/apm-utils'; +import { Logger } from '@kbn/core/server'; +import { compact } from 'lodash'; +import { + IStorageAdapter, + StorageAdapterBulkOperation, + StorageDocumentOf, + StorageSettings, +} from '.'; +import { ObservabilityESSearchRequest } from '../client/create_observability_es_client'; + +type StorageBulkOperation = + | { + index: { document: Omit; _id?: string }; + } + | { delete: { _id: string } }; + +export class StorageClient { + constructor(private readonly storage: IStorageAdapter, logger: Logger) {} + + search>( + operationName: string, + request: TSearchRequest + ) { + return withSpan(operationName, () => + this.storage.search, Omit>( + request + ) + ); + } + + async index({ + id, + document, + }: { + id?: string; + document: Omit, '_id'>; + }) { + await this.storage.index, '_id'>>({ + document, + refresh: 'wait_for', + id, + }); + } + + async delete(id: string) { + const searchResponse = await this.storage.search({ + query: { + bool: { + filter: [ + { + term: { + id, + }, + }, + ], + }, + }, + }); + + const document = searchResponse.hits.hits[0]; + + let deleted: boolean = false; + + if (document) { + await this.storage.delete({ id, index: document._index }); + deleted = true; + } + + return { acknowledged: true, deleted }; + } + + async bulk(operations: Array>>) { + const result = await this.storage.bulk({ + refresh: 'wait_for', + operations: operations.flatMap((operation): StorageAdapterBulkOperation[] => { + if ('index' in operation) { + return [ + { + index: { + _id: operation.index._id, + }, + }, + operation.index.document, + ]; + } + + return [operation]; + }), + }); + + if (result.errors) { + const errors = compact( + result.items.map((item) => { + const error = Object.values(item).find((operation) => operation.error)?.error; + return error; + }) + ); + return { + errors, + }; + } + + return { + acknowledged: true, + }; + } +} diff --git a/x-pack/solutions/observability/packages/utils_server/es/storage/types.ts b/x-pack/solutions/observability/packages/utils_server/es/storage/types.ts new file mode 100644 index 0000000000000..0c79be6f6080d --- /dev/null +++ b/x-pack/solutions/observability/packages/utils_server/es/storage/types.ts @@ -0,0 +1,133 @@ +/* + * 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 { MappingProperty } from '@elastic/elasticsearch/lib/api/types'; +import { merge } from 'lodash'; + +type AllMappingPropertyType = Required['type']; + +type StorageMappingPropertyType = AllMappingPropertyType & + ( + | 'text' + | 'match_only_text' + | 'keyword' + | 'boolean' + | 'date' + | 'byte' + | 'float' + | 'double' + | 'long' + ); + +type WithOptions = T extends any + ? T & { + required?: boolean; + multi_value?: boolean; + enum?: string[]; + } + : never; + +export type StorageMappingProperty = WithOptions< + Extract +>; + +type MappingPropertyOf = Extract< + StorageMappingProperty, + { type: TType } +>; + +type MappingPropertyFactory< + TType extends StorageMappingPropertyType, + TDefaults extends Partial | undefined> +> = > | undefined>( + overrides?: TOverrides +) => MappingPropertyOf & Exclude & Exclude; + +function createFactory< + TType extends StorageMappingPropertyType, + TDefaults extends Partial> | undefined +>(type: TType, defaults?: TDefaults): MappingPropertyFactory; + +function createFactory( + type: StorageMappingPropertyType, + defaults?: Partial +) { + return (overrides: Partial) => { + return { + ...defaults, + ...overrides, + type, + }; + }; +} + +const baseTypes = { + keyword: createFactory('keyword', { ignore_above: 1024 }), + match_only_text: createFactory('match_only_text'), + text: createFactory('text'), + double: createFactory('double'), + long: createFactory('long'), + boolean: createFactory('boolean'), + date: createFactory('date', { format: 'strict_date_optional_time' }), + byte: createFactory('byte'), + float: createFactory('float'), +} satisfies { + [TKey in StorageMappingPropertyType]: MappingPropertyFactory; +}; + +function enumFactory< + TEnum extends string, + TOverrides extends Partial> | undefined +>( + enums: TEnum[], + overrides?: TOverrides +): MappingPropertyOf<'keyword'> & { enum: TEnum[] } & Exclude; + +function enumFactory(enums: string[], overrides?: Partial>) { + const nextOverrides = merge({ enum: enums }, overrides); + const prop = baseTypes.keyword(nextOverrides); + return prop; +} + +const types = { + ...baseTypes, + enum: enumFactory, +}; + +type PrimitiveOf = { + keyword: TProperty extends { enum: infer TEnums } + ? TEnums extends Array + ? TEnum + : never + : string; + match_only_text: string; + text: string; + boolean: boolean; + date: TProperty extends { format: 'strict_date_optional_time' } ? string : string | number; + double: number; + long: number; + byte: number; + float: number; +}[TProperty['type']]; + +type MaybeMultiValue = TProperty extends { + multi_value: true; +} + ? TPrimitive[] + : TPrimitive; +type MaybeRequired = TProperty extends { + required: true; +} + ? TPrimitive + : TPrimitive | undefined; + +export type StorageFieldTypeOf = MaybeRequired< + TProperty, + MaybeMultiValue> +>; + +export { types }; diff --git a/x-pack/solutions/observability/packages/utils_server/tsconfig.json b/x-pack/solutions/observability/packages/utils_server/tsconfig.json index 33d7e75322f00..1bb54e734f891 100644 --- a/x-pack/solutions/observability/packages/utils_server/tsconfig.json +++ b/x-pack/solutions/observability/packages/utils_server/tsconfig.json @@ -28,5 +28,6 @@ "@kbn/calculate-auto", "@kbn/utility-types", "@kbn/task-manager-plugin", + "@kbn/es-errors", ] } diff --git a/x-pack/solutions/observability/plugins/streams/common/assets.ts b/x-pack/solutions/observability/plugins/streams/common/assets.ts new file mode 100644 index 0000000000000..9353418f14618 --- /dev/null +++ b/x-pack/solutions/observability/plugins/streams/common/assets.ts @@ -0,0 +1,40 @@ +/* + * 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 { ValuesType } from 'utility-types'; + +export const ASSET_TYPES = { + Dashboard: 'dashboard' as const, + Rule: 'rule' as const, + Slo: 'slo' as const, +}; + +export type AssetType = ValuesType; + +export interface AssetLink { + assetType: TAssetType; + assetId: string; +} + +export type DashboardLink = AssetLink<'dashboard'>; +export type SloLink = AssetLink<'slo'>; +export type RuleLink = AssetLink<'rule'>; + +export interface Asset extends AssetLink { + label: string; + tags: string[]; +} + +export type DashboardAsset = Asset<'dashboard'>; +export type SloAsset = Asset<'slo'>; +export type RuleAsset = Asset<'rule'>; + +export interface AssetTypeToAssetMap { + [ASSET_TYPES.Dashboard]: DashboardAsset; + [ASSET_TYPES.Slo]: SloAsset; + [ASSET_TYPES.Rule]: RuleAsset; +} diff --git a/x-pack/solutions/observability/plugins/streams/common/index.ts b/x-pack/solutions/observability/plugins/streams/common/index.ts new file mode 100644 index 0000000000000..ccb0dcb1bcbb1 --- /dev/null +++ b/x-pack/solutions/observability/plugins/streams/common/index.ts @@ -0,0 +1,8 @@ +/* + * 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 { Asset, AssetType } from './assets'; diff --git a/x-pack/solutions/observability/plugins/streams/kibana.jsonc b/x-pack/solutions/observability/plugins/streams/kibana.jsonc index b9ce6ef68e27e..943bc6223de23 100644 --- a/x-pack/solutions/observability/plugins/streams/kibana.jsonc +++ b/x-pack/solutions/observability/plugins/streams/kibana.jsonc @@ -16,7 +16,8 @@ "encryptedSavedObjects", "usageCollection", "licensing", - "taskManager" + "taskManager", + "alerting" ], "optionalPlugins": [ "cloud", diff --git a/x-pack/solutions/observability/plugins/streams/server/lib/streams/assets/asset_client.ts b/x-pack/solutions/observability/plugins/streams/server/lib/streams/assets/asset_client.ts new file mode 100644 index 0000000000000..22e50af643bd2 --- /dev/null +++ b/x-pack/solutions/observability/plugins/streams/server/lib/streams/assets/asset_client.ts @@ -0,0 +1,408 @@ +/* + * 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 { SanitizedRule } from '@kbn/alerting-plugin/common'; +import { RulesClient } from '@kbn/alerting-plugin/server'; +import { SavedObject, SavedObjectsClientContract } from '@kbn/core/server'; +import { termQuery } from '@kbn/observability-utils-server/es/queries/term_query'; +import { StorageClient, StorageDocumentOf } from '@kbn/observability-utils-server/es/storage'; +import { keyBy } from 'lodash'; +import objectHash from 'object-hash'; +import pLimit from 'p-limit'; +import { + ASSET_TYPES, + Asset, + AssetLink, + AssetType, + DashboardAsset, + SloAsset, + RuleAsset, +} from '../../../../common/assets'; +import { ASSET_ENTITY_ID, ASSET_ENTITY_TYPE, ASSET_TYPE } from './fields'; +import { AssetStorageSettings } from './storage_settings'; + +function sloSavedObjectToAsset( + sloId: string, + savedObject: SavedObject<{ name: string; tags: string[] }> +): SloAsset { + return { + assetId: sloId, + label: savedObject.attributes.name, + tags: savedObject.attributes.tags.concat( + savedObject.references.filter((ref) => ref.type === 'tag').map((ref) => ref.id) + ), + assetType: 'slo', + }; +} + +function dashboardSavedObjectToAsset( + dashboardId: string, + savedObject: SavedObject<{ title: string }> +): DashboardAsset { + return { + assetId: dashboardId, + label: savedObject.attributes.title, + tags: savedObject.references.filter((ref) => ref.type === 'tag').map((ref) => ref.id), + assetType: 'dashboard', + }; +} + +function ruleToAsset(ruleId: string, rule: SanitizedRule): RuleAsset { + return { + assetType: 'rule', + assetId: ruleId, + label: rule.name, + tags: rule.tags, + }; +} + +function getAssetDocument({ + assetId, + entityId, + entityType, + assetType, +}: AssetLink & { entityId: string; entityType: string }): StorageDocumentOf { + const doc = { + 'asset.id': assetId, + 'asset.type': assetType, + 'entity.id': entityId, + 'entity.type': entityType, + }; + + return { + _id: objectHash(doc), + ...doc, + }; +} + +interface AssetBulkIndexOperation { + index: { asset: AssetLink }; +} +interface AssetBulkDeleteOperation { + delete: { asset: AssetLink }; +} + +export type AssetBulkOperation = AssetBulkIndexOperation | AssetBulkDeleteOperation; + +export class AssetClient { + constructor( + private readonly clients: { + storageClient: StorageClient; + soClient: SavedObjectsClientContract; + rulesClient: RulesClient; + } + ) {} + + async linkAsset( + properties: { + entityId: string; + entityType: string; + } & AssetLink + ) { + const { _id: id, ...document } = getAssetDocument(properties); + + await this.clients.storageClient.index({ + id, + document, + }); + } + + async syncAssetList({ + entityId, + entityType, + assetType, + assetIds, + }: { + entityId: string; + entityType: string; + assetType: AssetType; + assetIds: string[]; + }) { + const assetsResponse = await this.clients.storageClient.search('get_assets_for_entity', { + size: 10_000, + track_total_hits: false, + query: { + bool: { + filter: [ + ...termQuery(ASSET_ENTITY_ID, entityId), + ...termQuery(ASSET_ENTITY_TYPE, entityType), + ...termQuery(ASSET_TYPE, assetType), + ], + }, + }, + }); + + const existingAssetLinks = assetsResponse.hits.hits.map((hit) => hit._source); + + const newAssetIds = assetIds.filter( + (assetId) => + !existingAssetLinks.some((existingAssetLink) => existingAssetLink['asset.id'] === assetId) + ); + + const assetIdsToRemove = existingAssetLinks + .map((existingAssetLink) => existingAssetLink['asset.id']) + .filter((assetId) => !assetIds.includes(assetId)); + + await Promise.all([ + ...newAssetIds.map((assetId) => + this.linkAsset({ + entityId, + entityType, + assetId, + assetType, + }) + ), + ...assetIdsToRemove.map((assetId) => + this.unlinkAsset({ + entityId, + entityType, + assetId, + assetType, + }) + ), + ]); + } + + async unlinkAsset( + properties: { + entityId: string; + entityType: string; + } & AssetLink + ) { + const { _id: id } = getAssetDocument(properties); + + await this.clients.storageClient.delete(id); + } + + async getAssetIds({ + entityId, + entityType, + assetType, + }: { + entityId: string; + entityType: 'stream'; + assetType: AssetType; + }): Promise { + const assetsResponse = await this.clients.storageClient.search('get_assets_for_entity', { + size: 10_000, + track_total_hits: false, + query: { + bool: { + filter: [ + ...termQuery(ASSET_ENTITY_ID, entityId), + ...termQuery(ASSET_ENTITY_TYPE, entityType), + ...termQuery(ASSET_TYPE, assetType), + ], + }, + }, + }); + + return assetsResponse.hits.hits.map((hit) => hit._source['asset.id']); + } + + async bulk( + { entityId, entityType }: { entityId: string; entityType: string }, + operations: AssetBulkOperation[] + ) { + return await this.clients.storageClient.bulk( + operations.map((operation) => { + const { _id, ...document } = getAssetDocument({ + ...Object.values(operation)[0].asset, + entityId, + entityType, + }); + + if ('index' in operation) { + return { + index: { + document, + _id, + }, + }; + } + + return { + delete: { + _id, + }, + }; + }) + ); + } + + async getAssets({ + entityId, + entityType, + }: { + entityId: string; + entityType: 'stream'; + }): Promise { + const assetsResponse = await this.clients.storageClient.search('get_assets_for_entity', { + size: 10_000, + track_total_hits: false, + query: { + bool: { + filter: [ + ...termQuery(ASSET_ENTITY_ID, entityId), + ...termQuery(ASSET_ENTITY_TYPE, entityType), + ], + }, + }, + }); + + const assetLinks = assetsResponse.hits.hits.map((hit) => hit._source); + + if (!assetLinks.length) { + return []; + } + + const idsByType = Object.fromEntries( + Object.values(ASSET_TYPES).map((type) => [type, [] as string[]]) + ) as Record; + + assetLinks.forEach((assetLink) => { + const assetType = assetLink['asset.type']; + const assetId = assetLink['asset.id']; + idsByType[assetType].push(assetId); + }); + + const limiter = pLimit(10); + + const [dashboards, rules, slos] = await Promise.all([ + idsByType.dashboard.length + ? this.clients.soClient + .bulkGet<{ title: string }>( + idsByType.dashboard.map((dashboardId) => ({ type: 'dashboard', id: dashboardId })) + ) + .then((response) => { + const dashboardsById = keyBy(response.saved_objects, 'id'); + + return idsByType.dashboard.flatMap((dashboardId): Asset[] => { + const dashboard = dashboardsById[dashboardId]; + if (dashboard && !dashboard.error) { + return [dashboardSavedObjectToAsset(dashboardId, dashboard)]; + } + return []; + }); + }) + : [], + Promise.all( + idsByType.rule.map((ruleId) => { + return limiter(() => + this.clients.rulesClient.get({ id: ruleId }).then((rule): Asset => { + return ruleToAsset(ruleId, rule); + }) + ); + }) + ), + idsByType.slo.length + ? this.clients.soClient + .find<{ name: string; tags: string[] }>({ + type: 'slo', + filter: `slo.attributes.id:(${idsByType.slo + .map((sloId) => `"${sloId}"`) + .join(' OR ')})`, + perPage: idsByType.slo.length, + }) + .then((soResponse) => { + const sloDefinitionsById = keyBy(soResponse.saved_objects, 'slo.attributes.id'); + + return idsByType.slo.flatMap((sloId): Asset[] => { + const sloDefinition = sloDefinitionsById[sloId]; + if (sloDefinition && !sloDefinition.error) { + return [sloSavedObjectToAsset(sloId, sloDefinition)]; + } + return []; + }); + }) + : [], + ]); + + return [...dashboards, ...rules, ...slos]; + } + + async getSuggestions({ + query, + assetTypes, + tags, + }: { + query: string; + assetTypes?: AssetType[]; + tags?: string[]; + }): Promise<{ hasMore: boolean; assets: Asset[] }> { + const perPage = 101; + + const searchAll = !assetTypes; + + const searchDashboardsOrSlos = + searchAll || assetTypes.includes('dashboard') || assetTypes.includes('slo'); + + const searchRules = searchAll || assetTypes.includes('rule'); + + const [suggestionsFromSlosAndDashboards, suggestionsFromRules] = await Promise.all([ + searchDashboardsOrSlos + ? this.clients.soClient + .find({ + type: ['dashboard' as const, 'slo' as const].filter( + (type) => searchAll || assetTypes.includes(type) + ), + search: query, + perPage, + ...(tags + ? { + hasReferenceOperator: 'OR', + hasReference: tags.map((tag) => ({ type: 'tag', id: tag })), + } + : {}), + }) + .then((results) => { + return results.saved_objects.map((savedObject) => { + if (savedObject.type === 'slo') { + const sloSavedObject = savedObject as SavedObject<{ + id: string; + name: string; + tags: string[]; + }>; + return sloSavedObjectToAsset(sloSavedObject.attributes.id, sloSavedObject); + } + + const dashboardSavedObject = savedObject as SavedObject<{ + title: string; + }>; + + return dashboardSavedObjectToAsset(dashboardSavedObject.id, dashboardSavedObject); + }); + }) + : Promise.resolve([]), + searchRules + ? this.clients.rulesClient + .find({ + options: { + perPage, + ...(tags + ? { + hasReferenceOperator: 'OR', + hasReference: tags.map((tag) => ({ type: 'tag', id: tag })), + } + : {}), + }, + }) + .then((results) => { + return results.data.map((rule) => { + return ruleToAsset(rule.id, rule); + }); + }) + : Promise.resolve([]), + ]); + + return { + assets: [...suggestionsFromRules, ...suggestionsFromSlosAndDashboards], + hasMore: + Math.max(suggestionsFromSlosAndDashboards.length, suggestionsFromRules.length) > + perPage - 1, + }; + } +} diff --git a/x-pack/solutions/observability/plugins/streams/server/lib/streams/assets/asset_service.ts b/x-pack/solutions/observability/plugins/streams/server/lib/streams/assets/asset_service.ts new file mode 100644 index 0000000000000..c83418aeda743 --- /dev/null +++ b/x-pack/solutions/observability/plugins/streams/server/lib/streams/assets/asset_service.ts @@ -0,0 +1,47 @@ +/* + * 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 { CoreSetup, KibanaRequest, Logger } from '@kbn/core/server'; +import { StorageIndexAdapter } from '@kbn/observability-utils-server/es/storage'; +import { Observable, defer, from, lastValueFrom, shareReplay } from 'rxjs'; +import { StreamsPluginStartDependencies } from '../../../types'; +import { AssetClient } from './asset_client'; +import { assetStorageSettings } from './storage_settings'; + +export class AssetService { + private adapter$: Observable>; + constructor( + private readonly coreSetup: CoreSetup, + private readonly logger: Logger + ) { + this.adapter$ = defer(() => from(this.getAdapter())).pipe(shareReplay(1)); + } + + async getAdapter(): Promise> { + const [coreStart] = await this.coreSetup.getStartServices(); + const esClient = coreStart.elasticsearch.client.asInternalUser; + + const adapter = new StorageIndexAdapter( + esClient, + this.logger.get('assets'), + assetStorageSettings + ); + + return adapter; + } + + async getClientWithRequest({ request }: { request: KibanaRequest }): Promise { + const [coreStart, pluginsStart] = await this.coreSetup.getStartServices(); + + const adapter = await lastValueFrom(this.adapter$); + return new AssetClient({ + storageClient: adapter.getClient(), + soClient: coreStart.savedObjects.getScopedClient(request), + rulesClient: await pluginsStart.alerting.getRulesClientWithRequest(request), + }); + } +} diff --git a/x-pack/solutions/observability/plugins/streams/server/lib/streams/assets/fields.ts b/x-pack/solutions/observability/plugins/streams/server/lib/streams/assets/fields.ts new file mode 100644 index 0000000000000..9dc57594829f7 --- /dev/null +++ b/x-pack/solutions/observability/plugins/streams/server/lib/streams/assets/fields.ts @@ -0,0 +1,11 @@ +/* + * 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 const ASSET_ENTITY_ID = 'entity.id'; +export const ASSET_ENTITY_TYPE = 'entity.type'; +export const ASSET_ASSET_ID = 'asset.id'; +export const ASSET_TYPE = 'asset.type'; diff --git a/x-pack/solutions/observability/plugins/streams/server/lib/streams/assets/storage_settings.ts b/x-pack/solutions/observability/plugins/streams/server/lib/streams/assets/storage_settings.ts new file mode 100644 index 0000000000000..92ac034c45353 --- /dev/null +++ b/x-pack/solutions/observability/plugins/streams/server/lib/streams/assets/storage_settings.ts @@ -0,0 +1,24 @@ +/* + * 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 { IndexStorageSettings, types } from '@kbn/observability-utils-server/es/storage'; +import { ASSET_ASSET_ID, ASSET_ENTITY_ID, ASSET_ENTITY_TYPE, ASSET_TYPE } from './fields'; +import { ASSET_TYPES } from '../../../../common/assets'; + +export const assetStorageSettings = { + name: '.kibana_streams_assets', + schema: { + properties: { + [ASSET_ASSET_ID]: types.keyword({ required: true }), + [ASSET_TYPE]: types.enum(Object.values(ASSET_TYPES), { required: true }), + [ASSET_ENTITY_ID]: types.keyword(), + [ASSET_ENTITY_TYPE]: types.keyword(), + }, + }, +} satisfies IndexStorageSettings; + +export type AssetStorageSettings = typeof assetStorageSettings; diff --git a/x-pack/solutions/observability/plugins/streams/server/lib/streams/stream_crud.ts b/x-pack/solutions/observability/plugins/streams/server/lib/streams/stream_crud.ts index 5669a3301e208..4d01e9d38fd27 100644 --- a/x-pack/solutions/observability/plugins/streams/server/lib/streams/stream_crud.ts +++ b/x-pack/solutions/observability/plugins/streams/server/lib/streams/stream_crud.ts @@ -44,6 +44,7 @@ import { upsertIngestPipeline, } from './ingest_pipelines/manage_ingest_pipelines'; import { getProcessingPipelineName, getReroutePipelineName } from './ingest_pipelines/name'; +import { AssetClient } from './assets/asset_client'; interface BaseParams { scopedClusterClient: IScopedClusterClient; @@ -56,6 +57,7 @@ interface BaseParamsWithDefinition extends BaseParams { interface DeleteStreamParams extends BaseParams { id: string; logger: Logger; + assetClient: AssetClient; } export async function deleteUnmanagedStreamObjects({ @@ -115,7 +117,12 @@ export async function deleteUnmanagedStreamObjects({ } } -export async function deleteStreamObjects({ id, scopedClusterClient, logger }: DeleteStreamParams) { +export async function deleteStreamObjects({ + id, + scopedClusterClient, + logger, + assetClient, +}: DeleteStreamParams) { await deleteDataStream({ esClient: scopedClusterClient.asCurrentUser, name: id, @@ -141,6 +148,12 @@ export async function deleteStreamObjects({ id, scopedClusterClient, logger }: D id: getReroutePipelineName(id), logger, }); + await assetClient.syncAssetList({ + entityId: id, + entityType: 'stream', + assetType: 'dashboard', + assetIds: [], + }); await scopedClusterClient.asInternalUser.delete({ id, index: STREAMS_INDEX, @@ -148,7 +161,10 @@ export async function deleteStreamObjects({ id, scopedClusterClient, logger }: D }); } -async function upsertInternalStream({ definition, scopedClusterClient }: BaseParamsWithDefinition) { +async function upsertInternalStream({ + definition: { dashboards, ...definition }, + scopedClusterClient, +}: BaseParamsWithDefinition) { return scopedClusterClient.asInternalUser.index({ id: definition.name, index: STREAMS_INDEX, @@ -157,41 +173,71 @@ async function upsertInternalStream({ definition, scopedClusterClient }: BasePar }); } +async function syncAssets({ + definition, + assetClient, +}: { + definition: StreamDefinition; + assetClient: AssetClient; +}) { + await assetClient.syncAssetList({ + entityId: definition.name, + entityType: 'stream', + assetType: 'dashboard', + assetIds: definition.dashboards ?? [], + }); +} + type ListStreamsParams = BaseParams; export async function listStreams({ scopedClusterClient, }: ListStreamsParams): Promise { - const response = await scopedClusterClient.asInternalUser.search({ + const [managedStreams, unmanagedStreams] = await Promise.all([ + listManagedStreams({ scopedClusterClient }), + listDataStreamsAsStreams({ scopedClusterClient }), + ]); + + const allDefinitionsById = new Map(managedStreams.map((stream) => [stream.name, stream])); + + unmanagedStreams.forEach((stream) => { + if (!allDefinitionsById.get(stream.name)) { + allDefinitionsById.set(stream.name, stream); + } + }); + + return { + streams: Array.from(allDefinitionsById.values()), + }; +} + +async function listManagedStreams({ + scopedClusterClient, +}: ListStreamsParams): Promise { + const streamsSearchResponse = await scopedClusterClient.asInternalUser.search({ index: STREAMS_INDEX, size: 10000, sort: [{ name: 'asc' }], }); - const dataStreams = await listDataStreamsAsStreams({ scopedClusterClient }); - let definitions = response.hits.hits.map((hit) => ({ ...hit._source! })); - const hasAccess = await Promise.all( - definitions.map((definition) => checkReadAccess({ id: definition.name, scopedClusterClient })) - ); - definitions = definitions.filter((_, index) => hasAccess[index]); - const definitionMap = new Map( - definitions.map((definition) => [definition.name, definition]) - ); - dataStreams.forEach((dataStream) => { - if (!definitionMap.has(dataStream.name)) { - definitionMap.set(dataStream.name, dataStream); - } + const streams = streamsSearchResponse.hits.hits.map((hit) => ({ + ...hit._source!, + managed: true, + })); + + const privileges = await scopedClusterClient.asCurrentUser.security.hasPrivileges({ + index: [{ names: streams.map((stream) => stream.name), privileges: ['read'] }], }); - return { - streams: Array.from(definitionMap.values()), - }; + return streams.filter((stream) => { + return privileges.index[stream.name]?.read === true; + }); } export async function listDataStreamsAsStreams({ scopedClusterClient, }: ListStreamsParams): Promise { - const response = await scopedClusterClient.asInternalUser.indices.getDataStream(); + const response = await scopedClusterClient.asCurrentUser.indices.getDataStream(); return response.data_streams .filter((dataStream) => dataStream.template.endsWith('@stream') === false) .map((dataStream) => ({ @@ -222,7 +268,7 @@ export async function readStream({ }); const definition = response._source as StreamDefinition; if (!skipAccessCheck) { - const hasAccess = await checkReadAccess({ id, scopedClusterClient }); + const hasAccess = await checkAccess({ id, scopedClusterClient }); if (!hasAccess) { throw new DefinitionNotFound(`Stream definition for ${id} not found.`); } @@ -263,21 +309,26 @@ async function getUnmanagedElasticsearchAssets({ name, scopedClusterClient, }: ReadUnmanagedAssetsParams) { - let dataStream: IndicesDataStream; + let dataStream: IndicesDataStream | undefined; try { - const response = await scopedClusterClient.asInternalUser.indices.getDataStream({ name }); + const response = await scopedClusterClient.asCurrentUser.indices.getDataStream({ name }); dataStream = response.data_streams[0]; } catch (e) { if (e.meta?.statusCode === 404) { - throw new DefinitionNotFound(`Stream definition for ${name} not found.`); + // fall through and throw not found + } else { + throw e; } - throw e; + } + + if (!dataStream) { + throw new DefinitionNotFound(`Stream definition for ${name} not found.`); } // retrieve linked index template, component template and ingest pipeline const templateName = dataStream.template; const componentTemplates: string[] = []; - const template = await scopedClusterClient.asInternalUser.indices.getIndexTemplate({ + const template = await scopedClusterClient.asCurrentUser.indices.getIndexTemplate({ name: templateName, }); if (template.index_templates.length) { @@ -286,7 +337,7 @@ async function getUnmanagedElasticsearchAssets({ }); } const writeIndexName = dataStream.indices.at(-1)?.index_name!; - const currentIndex = await scopedClusterClient.asInternalUser.indices.get({ + const currentIndex = await scopedClusterClient.asCurrentUser.indices.get({ index: writeIndexName, }); const ingestPipelineId = currentIndex[writeIndexName].settings?.index?.default_pipeline!; @@ -432,23 +483,43 @@ export async function checkStreamExists({ id, scopedClusterClient }: ReadStreamP } } -interface CheckReadAccessParams extends BaseParams { +interface CheckAccessParams extends BaseParams { id: string; } -export async function checkReadAccess({ +export async function checkAccess({ id, scopedClusterClient, -}: CheckReadAccessParams): Promise { - try { - return await scopedClusterClient.asCurrentUser.indices.exists({ index: id }); - } catch (e) { - return false; - } +}: CheckAccessParams): Promise<{ read: boolean; write: boolean }> { + return checkAccessBulk({ + ids: [id], + scopedClusterClient, + }).then((privileges) => privileges[id]); +} + +interface CheckAccessBulkParams extends BaseParams { + ids: string[]; } +export async function checkAccessBulk({ + ids, + scopedClusterClient, +}: CheckAccessBulkParams): Promise> { + const hasPrivilegesResponse = await scopedClusterClient.asCurrentUser.security.hasPrivileges({ + index: [{ names: ids, privileges: ['read', 'write'] }], + }); + + return Object.fromEntries( + ids.map((id) => { + const hasReadAccess = hasPrivilegesResponse.index[id].read === true; + const hasWriteAccess = hasPrivilegesResponse.index[id].write === true; + return [id, { read: hasReadAccess, write: hasWriteAccess }]; + }) + ); +} interface SyncStreamParams { scopedClusterClient: IScopedClusterClient; + assetClient: AssetClient; definition: StreamDefinition; rootDefinition?: StreamDefinition; logger: Logger; @@ -456,12 +527,13 @@ interface SyncStreamParams { export async function syncStream({ scopedClusterClient, + assetClient, definition, rootDefinition, logger, }: SyncStreamParams) { if (!isWiredStream(definition)) { - await syncUnmanagedStream({ scopedClusterClient, definition, logger }); + await syncUnmanagedStream({ scopedClusterClient, definition, logger, assetClient }); await upsertInternalStream({ scopedClusterClient, definition, @@ -512,6 +584,10 @@ export async function syncStream({ scopedClusterClient, definition, }); + await syncAssets({ + definition, + assetClient, + }); await rolloverDataStreamIfNecessary({ esClient: scopedClusterClient.asCurrentUser, name: definition.name, diff --git a/x-pack/solutions/observability/plugins/streams/server/plugin.ts b/x-pack/solutions/observability/plugins/streams/server/plugin.ts index 937f8c22b5be0..a40f1a9f2af2a 100644 --- a/x-pack/solutions/observability/plugins/streams/server/plugin.ts +++ b/x-pack/solutions/observability/plugins/streams/server/plugin.ts @@ -22,6 +22,7 @@ import { StreamsPluginStartDependencies, StreamsServer, } from './types'; +import { AssetService } from './lib/streams/assets/asset_service'; // eslint-disable-next-line @typescript-eslint/no-empty-interface export interface StreamsPluginSetup {} @@ -51,21 +52,28 @@ export class StreamsPlugin this.logger = context.logger.get(); } - public setup(core: CoreSetup, plugins: StreamsPluginSetupDependencies): StreamsPluginSetup { + public setup( + core: CoreSetup, + plugins: StreamsPluginSetupDependencies + ): StreamsPluginSetup { this.server = { config: this.config, logger: this.logger, } as StreamsServer; + const assetService = new AssetService(core, this.logger); + registerRoutes({ repository: streamsRouteRepository, dependencies: { + assets: assetService, server: this.server, getScopedClients: async ({ request }: { request: KibanaRequest }) => { const [coreStart] = await core.getStartServices(); + const assetClient = await assetService.getClientWithRequest({ request }); const scopedClusterClient = coreStart.elasticsearch.client.asScoped(request); const soClient = coreStart.savedObjects.getScopedClient(request); - return { scopedClusterClient, soClient }; + return { scopedClusterClient, soClient, assetClient }; }, }, core, diff --git a/x-pack/solutions/observability/plugins/streams/server/routes/dashboards/route.ts b/x-pack/solutions/observability/plugins/streams/server/routes/dashboards/route.ts new file mode 100644 index 0000000000000..b5ea8646daf6b --- /dev/null +++ b/x-pack/solutions/observability/plugins/streams/server/routes/dashboards/route.ts @@ -0,0 +1,262 @@ +/* + * 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 { z } from '@kbn/zod'; +import { ErrorCause } from '@elastic/elasticsearch/lib/api/types'; +import { internal } from '@hapi/boom'; +import { Asset, DashboardAsset } from '../../../common/assets'; +import { createServerRoute } from '../create_server_route'; + +export interface SanitizedDashboardAsset { + id: string; + label: string; + tags: string[]; +} + +export interface ListDashboardsResponse { + dashboards: SanitizedDashboardAsset[]; +} + +export interface LinkDashboardResponse { + acknowledged: boolean; +} + +export interface UnlinkDashboardResponse { + acknowledged: boolean; +} + +export interface SuggestDashboardResponse { + suggestions: SanitizedDashboardAsset[]; +} + +export type BulkUpdateAssetsResponse = + | { + acknowledged: boolean; + } + | { errors: ErrorCause[] }; + +function sanitizeDashboardAsset(asset: DashboardAsset): SanitizedDashboardAsset { + return { + id: asset.assetId, + label: asset.label, + tags: asset.tags, + }; +} + +const listDashboardsRoute = createServerRoute({ + endpoint: 'GET /api/streams/{id}/dashboards', + options: { + access: 'internal', + }, + params: z.object({ + path: z.object({ + id: z.string(), + }), + }), + async handler({ params, request, assets }): Promise { + const assetsClient = await assets.getClientWithRequest({ request }); + + const { + path: { id: streamId }, + } = params; + + function isDashboard(asset: Asset): asset is DashboardAsset { + return asset.assetType === 'dashboard'; + } + + return { + dashboards: ( + await assetsClient.getAssets({ + entityId: streamId, + entityType: 'stream', + }) + ) + .filter(isDashboard) + .map(sanitizeDashboardAsset), + }; + }, +}); + +const linkDashboardRoute = createServerRoute({ + endpoint: 'PUT /api/streams/{id}/dashboards/{dashboardId}', + options: { + access: 'internal', + }, + params: z.object({ + path: z.object({ + id: z.string(), + dashboardId: z.string(), + }), + }), + handler: async ({ params, request, assets }): Promise => { + const assetsClient = await assets.getClientWithRequest({ request }); + + const { + path: { dashboardId, id: streamId }, + } = params; + + await assetsClient.linkAsset({ + entityId: streamId, + entityType: 'stream', + assetId: dashboardId, + assetType: 'dashboard', + }); + + return { + acknowledged: true, + }; + }, +}); + +const unlinkDashboardRoute = createServerRoute({ + endpoint: 'DELETE /api/streams/{id}/dashboards/{dashboardId}', + options: { + access: 'internal', + }, + params: z.object({ + path: z.object({ + id: z.string(), + dashboardId: z.string(), + }), + }), + handler: async ({ params, request, assets }): Promise => { + const assetsClient = await assets.getClientWithRequest({ request }); + + const { + path: { dashboardId, id: streamId }, + } = params; + + await assetsClient.unlinkAsset({ + entityId: streamId, + entityType: 'stream', + assetId: dashboardId, + assetType: 'dashboard', + }); + + return { + acknowledged: true, + }; + }, +}); + +const suggestDashboardsRoute = createServerRoute({ + endpoint: 'POST /api/streams/{id}/dashboards/_suggestions', + options: { + access: 'internal', + }, + params: z.object({ + path: z.object({ + id: z.string(), + }), + query: z.object({ + query: z.string(), + }), + body: z.object({ + tags: z.optional(z.array(z.string())), + }), + }), + handler: async ({ params, request, assets }): Promise => { + const assetsClient = await assets.getClientWithRequest({ request }); + + const { + query: { query }, + body: { tags }, + } = params; + + const suggestions = ( + await assetsClient.getSuggestions({ + assetTypes: ['dashboard'], + query, + tags, + }) + ).assets.map((asset) => { + return sanitizeDashboardAsset(asset as DashboardAsset); + }); + + return { + suggestions, + }; + }, +}); + +const dashboardSchema = z.object({ + id: z.string(), +}); + +const bulkDashboardsRoute = createServerRoute({ + endpoint: `POST /api/streams/{id}/dashboards/_bulk`, + options: { + access: 'internal', + }, + params: z.object({ + path: z.object({ + id: z.string(), + }), + body: z.object({ + operations: z.array( + z.union([ + z.object({ + index: dashboardSchema, + }), + z.object({ + delete: dashboardSchema, + }), + ]) + ), + }), + }), + handler: async ({ params, request, assets, logger }): Promise => { + const assetsClient = await assets.getClientWithRequest({ request }); + + const { + path: { id: streamId }, + body: { operations }, + } = params; + + const result = await assetsClient.bulk( + { + entityId: streamId, + entityType: 'stream', + }, + operations.map((operation) => { + if ('index' in operation) { + return { + index: { + asset: { + assetType: 'dashboard', + assetId: operation.index.id, + }, + }, + }; + } + return { + delete: { + asset: { + assetType: 'dashboard', + assetId: operation.delete.id, + }, + }, + }; + }) + ); + + if (result.errors) { + logger.error(`Error indexing ${result.errors.length} items`); + throw internal(`Could not index all items`, { errors: result.errors }); + } + + return { acknowledged: true }; + }, +}); + +export const dashboardRoutes = { + ...listDashboardsRoute, + ...linkDashboardRoute, + ...unlinkDashboardRoute, + ...suggestDashboardsRoute, + ...bulkDashboardsRoute, +}; diff --git a/x-pack/solutions/observability/plugins/streams/server/routes/index.ts b/x-pack/solutions/observability/plugins/streams/server/routes/index.ts index e6c53e33e217e..fe10f3e282c46 100644 --- a/x-pack/solutions/observability/plugins/streams/server/routes/index.ts +++ b/x-pack/solutions/observability/plugins/streams/server/routes/index.ts @@ -5,6 +5,7 @@ * 2.0. */ +import { dashboardRoutes } from './dashboards/route'; import { esqlRoutes } from './esql/route'; import { deleteStreamRoute } from './streams/delete'; import { disableStreamsRoute } from './streams/disable'; @@ -30,6 +31,7 @@ export const streamsRouteRepository = { ...streamsStatusRoutes, ...esqlRoutes, ...disableStreamsRoute, + ...dashboardRoutes, ...sampleStreamRoute, ...unmappedFieldsRoute, ...schemaFieldsSimulationRoute, diff --git a/x-pack/solutions/observability/plugins/streams/server/routes/streams/delete.ts b/x-pack/solutions/observability/plugins/streams/server/routes/streams/delete.ts index 698d0f7f81d38..cc773523d9719 100644 --- a/x-pack/solutions/observability/plugins/streams/server/routes/streams/delete.ts +++ b/x-pack/solutions/observability/plugins/streams/server/routes/streams/delete.ts @@ -25,6 +25,7 @@ import { } from '../../lib/streams/stream_crud'; import { MalformedStreamId } from '../../lib/streams/errors/malformed_stream_id'; import { getParentId } from '../../lib/streams/helpers/hierarchy'; +import { AssetClient } from '../../lib/streams/assets/asset_client'; export const deleteStreamRoute = createServerRoute({ endpoint: 'DELETE /api/streams/{id}', @@ -50,9 +51,21 @@ export const deleteStreamRoute = createServerRoute({ getScopedClients, }): Promise<{ acknowledged: true }> => { try { - const { scopedClusterClient } = await getScopedClients({ request }); + const { scopedClusterClient, assetClient } = await getScopedClients({ request }); - await deleteStream(scopedClusterClient, params.path.id, logger); + const parentId = getParentId(params.path.id); + if (parentId) { + // need to update parent first to cut off documents streaming down + await updateParentStream( + scopedClusterClient, + assetClient, + params.path.id, + parentId, + logger + ); + } + + await deleteStream(scopedClusterClient, assetClient, params.path.id, logger); return { acknowledged: true }; } catch (e) { @@ -75,13 +88,14 @@ export const deleteStreamRoute = createServerRoute({ export async function deleteStream( scopedClusterClient: IScopedClusterClient, + assetClient: AssetClient, id: string, logger: Logger ) { try { const definition = await readStream({ scopedClusterClient, id }); if (!isWiredStream(definition)) { - await deleteUnmanagedStreamObjects({ scopedClusterClient, id, logger }); + await deleteUnmanagedStreamObjects({ scopedClusterClient, id, logger, assetClient }); return; } @@ -91,11 +105,11 @@ export async function deleteStream( } // need to update parent first to cut off documents streaming down - await updateParentStream(scopedClusterClient, id, parentId, logger); + await updateParentStream(scopedClusterClient, assetClient, id, parentId, logger); for (const child of definition.stream.ingest.routing) { - await deleteStream(scopedClusterClient, child.name, logger); + await deleteStream(scopedClusterClient, assetClient, child.name, logger); } - await deleteStreamObjects({ scopedClusterClient, id, logger }); + await deleteStreamObjects({ scopedClusterClient, id, logger, assetClient }); } catch (e) { if (e instanceof DefinitionNotFound) { logger.debug(`Stream definition for ${id} not found.`); @@ -107,6 +121,7 @@ export async function deleteStream( async function updateParentStream( scopedClusterClient: IScopedClusterClient, + assetClient: AssetClient, id: string, parentId: string, logger: Logger @@ -122,6 +137,7 @@ async function updateParentStream( await syncStream({ scopedClusterClient, + assetClient, definition: parentDefinition, logger, }); diff --git a/x-pack/solutions/observability/plugins/streams/server/routes/streams/disable.ts b/x-pack/solutions/observability/plugins/streams/server/routes/streams/disable.ts index 3cf369f6da76d..57519e2ce4be5 100644 --- a/x-pack/solutions/observability/plugins/streams/server/routes/streams/disable.ts +++ b/x-pack/solutions/observability/plugins/streams/server/routes/streams/disable.ts @@ -24,9 +24,9 @@ export const disableStreamsRoute = createServerRoute({ }, handler: async ({ request, logger, getScopedClients }): Promise<{ acknowledged: true }> => { try { - const { scopedClusterClient } = await getScopedClients({ request }); + const { scopedClusterClient, assetClient } = await getScopedClients({ request }); - await deleteStream(scopedClusterClient, 'logs', logger); + await deleteStream(scopedClusterClient, assetClient, 'logs', logger); return { acknowledged: true }; } catch (e) { diff --git a/x-pack/solutions/observability/plugins/streams/server/routes/streams/edit.ts b/x-pack/solutions/observability/plugins/streams/server/routes/streams/edit.ts index cf88835602076..1ed24898f2d03 100644 --- a/x-pack/solutions/observability/plugins/streams/server/routes/streams/edit.ts +++ b/x-pack/solutions/observability/plugins/streams/server/routes/streams/edit.ts @@ -34,6 +34,7 @@ import { import { MalformedStreamId } from '../../lib/streams/errors/malformed_stream_id'; import { getParentId } from '../../lib/streams/helpers/hierarchy'; import { MalformedChildren } from '../../lib/streams/errors/malformed_children'; +import { AssetClient } from '../../lib/streams/assets/asset_client'; import { validateCondition } from '../../lib/streams/helpers/condition_fields'; export const editStreamRoute = createServerRoute({ @@ -56,7 +57,7 @@ export const editStreamRoute = createServerRoute({ }), handler: async ({ params, logger, request, getScopedClients }) => { try { - const { scopedClusterClient } = await getScopedClients({ request }); + const { scopedClusterClient, assetClient } = await getScopedClients({ request }); const streamDefinition: StreamDefinition = { stream: params.body, name: params.path.id }; if (!isWiredStream(streamDefinition)) { @@ -65,6 +66,7 @@ export const editStreamRoute = createServerRoute({ definition: streamDefinition, rootDefinition: undefined, logger, + assetClient, }); return { acknowledged: true }; } @@ -113,6 +115,7 @@ export const editStreamRoute = createServerRoute({ await syncStream({ scopedClusterClient, + assetClient, definition: childDefinition, logger, }); @@ -123,11 +126,13 @@ export const editStreamRoute = createServerRoute({ definition: { ...streamDefinition, name: params.path.id }, rootDefinition: parentDefinition, logger, + assetClient, }); if (parentId) { parentDefinition = await updateParentStream( scopedClusterClient, + assetClient, parentId, params.path.id, logger @@ -155,6 +160,7 @@ export const editStreamRoute = createServerRoute({ async function updateParentStream( scopedClusterClient: IScopedClusterClient, + assetClient: AssetClient, parentId: string, id: string, logger: Logger @@ -173,6 +179,7 @@ async function updateParentStream( await syncStream({ scopedClusterClient, + assetClient, definition: parentDefinition, logger, }); diff --git a/x-pack/solutions/observability/plugins/streams/server/routes/streams/enable.ts b/x-pack/solutions/observability/plugins/streams/server/routes/streams/enable.ts index 8b479813f87af..3ce58300683d5 100644 --- a/x-pack/solutions/observability/plugins/streams/server/routes/streams/enable.ts +++ b/x-pack/solutions/observability/plugins/streams/server/routes/streams/enable.ts @@ -32,7 +32,7 @@ export const enableStreamsRoute = createServerRoute({ getScopedClients, }): Promise<{ acknowledged: true; message: string }> => { try { - const { scopedClusterClient } = await getScopedClients({ request }); + const { scopedClusterClient, assetClient } = await getScopedClients({ request }); const alreadyEnabled = await streamsEnabled({ scopedClusterClient }); if (alreadyEnabled) { return { acknowledged: true, message: 'Streams was already enabled' }; @@ -40,6 +40,7 @@ export const enableStreamsRoute = createServerRoute({ await createStreamsIndex(scopedClusterClient); await syncStream({ scopedClusterClient, + assetClient, definition: rootStreamDefinition, logger, }); diff --git a/x-pack/solutions/observability/plugins/streams/server/routes/streams/fork.ts b/x-pack/solutions/observability/plugins/streams/server/routes/streams/fork.ts index 447fdfcc84978..8453045863794 100644 --- a/x-pack/solutions/observability/plugins/streams/server/routes/streams/fork.ts +++ b/x-pack/solutions/observability/plugins/streams/server/routes/streams/fork.ts @@ -51,7 +51,7 @@ export const forkStreamsRoute = createServerRoute({ validateCondition(params.body.condition); - const { scopedClusterClient } = await getScopedClients({ request }); + const { scopedClusterClient, assetClient } = await getScopedClients({ request }); const rootDefinition = await readStream({ scopedClusterClient, @@ -91,6 +91,7 @@ export const forkStreamsRoute = createServerRoute({ // need to create the child first, otherwise we risk streaming data even though the child data stream is not ready await syncStream({ scopedClusterClient, + assetClient, definition: childDefinition, rootDefinition, logger, @@ -103,6 +104,7 @@ export const forkStreamsRoute = createServerRoute({ await syncStream({ scopedClusterClient, + assetClient, definition: rootDefinition, rootDefinition, logger, diff --git a/x-pack/solutions/observability/plugins/streams/server/routes/streams/list.ts b/x-pack/solutions/observability/plugins/streams/server/routes/streams/list.ts index 66edc3c7954b4..8f2755ee73d72 100644 --- a/x-pack/solutions/observability/plugins/streams/server/routes/streams/list.ts +++ b/x-pack/solutions/observability/plugins/streams/server/routes/streams/list.ts @@ -28,7 +28,7 @@ export const listStreamsRoute = createServerRoute({ handler: async ({ request, getScopedClients }): Promise => { try { const { scopedClusterClient } = await getScopedClients({ request }); - return listStreams({ scopedClusterClient }); + return await listStreams({ scopedClusterClient }); } catch (e) { if (e instanceof DefinitionNotFound) { throw notFound(e); diff --git a/x-pack/solutions/observability/plugins/streams/server/routes/streams/read.ts b/x-pack/solutions/observability/plugins/streams/server/routes/streams/read.ts index cd3d43934f107..f7c967035bd68 100644 --- a/x-pack/solutions/observability/plugins/streams/server/routes/streams/read.ts +++ b/x-pack/solutions/observability/plugins/streams/server/routes/streams/read.ts @@ -9,9 +9,9 @@ import { z } from '@kbn/zod'; import { notFound, internal } from '@hapi/boom'; import { FieldDefinitionConfig, - isIngestStream, isWiredStream, ReadStreamDefinition, + WiredReadStreamDefinition, } from '@kbn/streams-schema'; import { createServerRoute } from '../create_server_route'; import { DefinitionNotFound } from '../../lib/streams/errors'; @@ -34,17 +34,21 @@ export const readStreamRoute = createServerRoute({ }), handler: async ({ params, request, getScopedClients }): Promise => { try { - const { scopedClusterClient } = await getScopedClients({ request }); + const { scopedClusterClient, assetClient } = await getScopedClients({ request }); const streamEntity = await readStream({ scopedClusterClient, id: params.path.id, }); + const dashboards = await assetClient.getAssetIds({ + entityId: streamEntity.name, + entityType: 'stream', + assetType: 'dashboard', + }); - // TODO: I have no idea why I can just do `isIngestStream` here but when I do, - // streamEntity becomes `streamEntity: never` in the statements afterwards - if (!isWiredStream(streamEntity) && isIngestStream(streamEntity)) { + if (!isWiredStream(streamEntity)) { return { ...streamEntity, + dashboards, inherited_fields: {}, }; } @@ -54,8 +58,9 @@ export const readStreamRoute = createServerRoute({ scopedClusterClient, }); - const body = { + const body: WiredReadStreamDefinition = { ...streamEntity, + dashboards, inherited_fields: ancestors.reduce((acc, def) => { Object.entries(def.stream.ingest.wired.fields).forEach(([key, fieldDef]) => { acc[key] = { ...fieldDef, from: def.name }; diff --git a/x-pack/solutions/observability/plugins/streams/server/routes/streams/resync.ts b/x-pack/solutions/observability/plugins/streams/server/routes/streams/resync.ts index 73955a2bd9bb5..eb3bad1db58d2 100644 --- a/x-pack/solutions/observability/plugins/streams/server/routes/streams/resync.ts +++ b/x-pack/solutions/observability/plugins/streams/server/routes/streams/resync.ts @@ -23,7 +23,7 @@ export const resyncStreamsRoute = createServerRoute({ }, params: z.object({}), handler: async ({ logger, request, getScopedClients }): Promise<{ acknowledged: true }> => { - const { scopedClusterClient } = await getScopedClients({ request }); + const { scopedClusterClient, assetClient } = await getScopedClients({ request }); const { streams } = await listStreams({ scopedClusterClient }); @@ -35,6 +35,7 @@ export const resyncStreamsRoute = createServerRoute({ await syncStream({ scopedClusterClient, + assetClient, definition, logger, }); diff --git a/x-pack/solutions/observability/plugins/streams/server/routes/streams/sample.ts b/x-pack/solutions/observability/plugins/streams/server/routes/streams/sample.ts index f912e2e27fd96..1e4508149f4e7 100644 --- a/x-pack/solutions/observability/plugins/streams/server/routes/streams/sample.ts +++ b/x-pack/solutions/observability/plugins/streams/server/routes/streams/sample.ts @@ -7,10 +7,11 @@ import { z } from '@kbn/zod'; import { notFound, internal } from '@hapi/boom'; +import { errors } from '@elastic/elasticsearch'; import { conditionSchema } from '@kbn/streams-schema'; import { createServerRoute } from '../create_server_route'; import { DefinitionNotFound } from '../../lib/streams/errors'; -import { checkReadAccess } from '../../lib/streams/stream_crud'; +import { checkAccess } from '../../lib/streams/stream_crud'; import { conditionToQueryDsl } from '../../lib/streams/helpers/condition_to_query_dsl'; import { getFields, isComplete } from '../../lib/streams/helpers/condition_fields'; @@ -39,8 +40,8 @@ export const sampleStreamRoute = createServerRoute({ try { const { scopedClusterClient } = await getScopedClients({ request }); - const hasAccess = await checkReadAccess({ id: params.path.id, scopedClusterClient }); - if (!hasAccess) { + const { read } = await checkAccess({ id: params.path.id, scopedClusterClient }); + if (!read) { throw new DefinitionNotFound(`Stream definition for ${params.path.id} not found.`); } const searchBody = { @@ -84,16 +85,20 @@ export const sampleStreamRoute = createServerRoute({ }; const results = await scopedClusterClient.asCurrentUser.search({ index: params.path.id, + allow_no_indices: true, ...searchBody, }); return { documents: results.hits.hits.map((hit) => hit._source) }; - } catch (e) { - if (e instanceof DefinitionNotFound) { - throw notFound(e); + } catch (error) { + if (error instanceof errors.ResponseError && error.meta.statusCode === 404) { + throw notFound(error); + } + if (error instanceof DefinitionNotFound) { + throw notFound(error); } - throw internal(e); + throw internal(error); } }, }); diff --git a/x-pack/solutions/observability/plugins/streams/server/routes/streams/schema/fields_simulation.ts b/x-pack/solutions/observability/plugins/streams/server/routes/streams/schema/fields_simulation.ts index 9db5a7013f01e..140a5ce7a3b29 100644 --- a/x-pack/solutions/observability/plugins/streams/server/routes/streams/schema/fields_simulation.ts +++ b/x-pack/solutions/observability/plugins/streams/server/routes/streams/schema/fields_simulation.ts @@ -11,7 +11,7 @@ import { getFlattenedObject } from '@kbn/std'; import { fieldDefinitionConfigSchema } from '@kbn/streams-schema'; import { createServerRoute } from '../../create_server_route'; import { DefinitionNotFound } from '../../../lib/streams/errors'; -import { checkReadAccess } from '../../../lib/streams/stream_crud'; +import { checkAccess } from '../../../lib/streams/stream_crud'; const SAMPLE_SIZE = 200; @@ -45,8 +45,8 @@ export const schemaFieldsSimulationRoute = createServerRoute({ try { const { scopedClusterClient } = await getScopedClients({ request }); - const hasAccess = await checkReadAccess({ id: params.path.id, scopedClusterClient }); - if (!hasAccess) { + const { read } = await checkAccess({ id: params.path.id, scopedClusterClient }); + if (!read) { throw new DefinitionNotFound(`Stream definition for ${params.path.id} not found.`); } diff --git a/x-pack/solutions/observability/plugins/streams/server/routes/streams/schema/unmapped_fields.ts b/x-pack/solutions/observability/plugins/streams/server/routes/streams/schema/unmapped_fields.ts index 12faa12f9cee4..79013f57d9ca5 100644 --- a/x-pack/solutions/observability/plugins/streams/server/routes/streams/schema/unmapped_fields.ts +++ b/x-pack/solutions/observability/plugins/streams/server/routes/streams/schema/unmapped_fields.ts @@ -10,7 +10,7 @@ import { internal, notFound } from '@hapi/boom'; import { getFlattenedObject } from '@kbn/std'; import { isWiredStream } from '@kbn/streams-schema'; import { DefinitionNotFound } from '../../../lib/streams/errors'; -import { checkReadAccess, readAncestors, readStream } from '../../../lib/streams/stream_crud'; +import { checkAccess, readAncestors, readStream } from '../../../lib/streams/stream_crud'; import { createServerRoute } from '../../create_server_route'; const SAMPLE_SIZE = 500; @@ -34,8 +34,8 @@ export const unmappedFieldsRoute = createServerRoute({ try { const { scopedClusterClient } = await getScopedClients({ request }); - const hasAccess = await checkReadAccess({ id: params.path.id, scopedClusterClient }); - if (!hasAccess) { + const { read } = await checkAccess({ id: params.path.id, scopedClusterClient }); + if (!read) { throw new DefinitionNotFound(`Stream definition for ${params.path.id} not found.`); } diff --git a/x-pack/solutions/observability/plugins/streams/server/routes/types.ts b/x-pack/solutions/observability/plugins/streams/server/routes/types.ts index d547d56c088cd..3a9f903120e73 100644 --- a/x-pack/solutions/observability/plugins/streams/server/routes/types.ts +++ b/x-pack/solutions/observability/plugins/streams/server/routes/types.ts @@ -10,12 +10,16 @@ import { DefaultRouteHandlerResources } from '@kbn/server-route-repository'; import { IScopedClusterClient } from '@kbn/core-elasticsearch-server'; import { SavedObjectsClientContract } from '@kbn/core-saved-objects-api-server'; import { StreamsServer } from '../types'; +import { AssetService } from '../lib/streams/assets/asset_service'; +import { AssetClient } from '../lib/streams/assets/asset_client'; export interface RouteDependencies { + assets: AssetService; server: StreamsServer; getScopedClients: ({ request }: { request: KibanaRequest }) => Promise<{ scopedClusterClient: IScopedClusterClient; soClient: SavedObjectsClientContract; + assetClient: AssetClient; }>; } diff --git a/x-pack/solutions/observability/plugins/streams/server/types.ts b/x-pack/solutions/observability/plugins/streams/server/types.ts index f119faa0ed010..63ed5328082a7 100644 --- a/x-pack/solutions/observability/plugins/streams/server/types.ts +++ b/x-pack/solutions/observability/plugins/streams/server/types.ts @@ -5,18 +5,19 @@ * 2.0. */ -import { CoreStart, ElasticsearchClient, Logger } from '@kbn/core/server'; -import { SecurityPluginStart } from '@kbn/security-plugin/server'; -import { +import type { CoreStart, ElasticsearchClient, Logger } from '@kbn/core/server'; +import type { SecurityPluginStart } from '@kbn/security-plugin/server'; +import type { EncryptedSavedObjectsPluginSetup, EncryptedSavedObjectsPluginStart, } from '@kbn/encrypted-saved-objects-plugin/server'; -import { LicensingPluginStart } from '@kbn/licensing-plugin/server'; -import { +import type { LicensingPluginStart } from '@kbn/licensing-plugin/server'; +import type { TaskManagerSetupContract, TaskManagerStartContract, } from '@kbn/task-manager-plugin/server'; -import { StreamsConfig } from '../common/config'; +import type { AlertingServerSetup, AlertingServerStart } from '@kbn/alerting-plugin/server'; +import type { StreamsConfig } from '../common/config'; export interface StreamsServer { core: CoreStart; @@ -35,6 +36,7 @@ export interface ElasticsearchAccessorOptions { export interface StreamsPluginSetupDependencies { encryptedSavedObjects: EncryptedSavedObjectsPluginSetup; taskManager: TaskManagerSetupContract; + alerting: AlertingServerSetup; } export interface StreamsPluginStartDependencies { @@ -42,4 +44,5 @@ export interface StreamsPluginStartDependencies { encryptedSavedObjects: EncryptedSavedObjectsPluginStart; licensing: LicensingPluginStart; taskManager: TaskManagerStartContract; + alerting: AlertingServerStart; } diff --git a/x-pack/solutions/observability/plugins/streams/tsconfig.json b/x-pack/solutions/observability/plugins/streams/tsconfig.json index 464f184918c96..27743fbd9f70c 100644 --- a/x-pack/solutions/observability/plugins/streams/tsconfig.json +++ b/x-pack/solutions/observability/plugins/streams/tsconfig.json @@ -30,6 +30,7 @@ "@kbn/server-route-repository-client", "@kbn/observability-utils-server", "@kbn/observability-utils-common", + "@kbn/alerting-plugin", "@kbn/std", "@kbn/safer-lodash-set", "@kbn/streams-schema" diff --git a/x-pack/solutions/observability/plugins/streams_app/.storybook/get_mock_streams_app_context.tsx b/x-pack/solutions/observability/plugins/streams_app/.storybook/get_mock_streams_app_context.tsx index cb3148a7f6644..8e5dad100fc31 100644 --- a/x-pack/solutions/observability/plugins/streams_app/.storybook/get_mock_streams_app_context.tsx +++ b/x-pack/solutions/observability/plugins/streams_app/.storybook/get_mock_streams_app_context.tsx @@ -13,6 +13,7 @@ import type { ObservabilitySharedPluginStart } from '@kbn/observability-shared-p import type { UnifiedSearchPublicPluginStart } from '@kbn/unified-search-plugin/public'; import type { SharePublicStart } from '@kbn/share-plugin/public/plugin'; import { NavigationPublicStart } from '@kbn/navigation-plugin/public/types'; +import type { SavedObjectTaggingPluginStart } from '@kbn/saved-objects-tagging-plugin/public'; import type { StreamsAppKibanaContext } from '../public/hooks/use_kibana'; export function getMockStreamsAppContext(): StreamsAppKibanaContext { @@ -29,6 +30,7 @@ export function getMockStreamsAppContext(): StreamsAppKibanaContext { streams: {} as unknown as StreamsPluginStart, share: {} as unknown as SharePublicStart, navigation: {} as unknown as NavigationPublicStart, + savedObjectsTagging: {} as unknown as SavedObjectTaggingPluginStart, }, }, services: { diff --git a/x-pack/solutions/observability/plugins/streams_app/kibana.jsonc b/x-pack/solutions/observability/plugins/streams_app/kibana.jsonc index 09f356f0654ba..dd1026e3aaf32 100644 --- a/x-pack/solutions/observability/plugins/streams_app/kibana.jsonc +++ b/x-pack/solutions/observability/plugins/streams_app/kibana.jsonc @@ -16,6 +16,7 @@ "dataViews", "unifiedSearch", "share", + "savedObjectsTagging", "navigation" ], "requiredBundles": [ diff --git a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_dashboards_view/add_dashboard_flyout.tsx b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_dashboards_view/add_dashboard_flyout.tsx new file mode 100644 index 0000000000000..4e3918748e467 --- /dev/null +++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_dashboards_view/add_dashboard_flyout.tsx @@ -0,0 +1,233 @@ +/* + * 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 { + EuiButton, + EuiFilterButton, + EuiFilterGroup, + EuiFlexGroup, + EuiFlexItem, + EuiFlyout, + EuiFlyoutBody, + EuiFlyoutFooter, + EuiFlyoutHeader, + EuiPopover, + EuiPopoverTitle, + EuiSearchBar, + EuiSelectable, + EuiText, + EuiTitle, + useGeneratedHtmlId, +} from '@elastic/eui'; +import { i18n } from '@kbn/i18n'; +import { debounce } from 'lodash'; +import React, { useMemo, useState, useEffect } from 'react'; +import type { SanitizedDashboardAsset } from '@kbn/streams-plugin/server/routes/dashboards/route'; +import { useKibana } from '../../hooks/use_kibana'; +import { useStreamsAppFetch } from '../../hooks/use_streams_app_fetch'; +import { DashboardsTable } from './dashboard_table'; + +export function AddDashboardFlyout({ + entityId, + onAddDashboards, + linkedDashboards, + onClose, +}: { + entityId: string; + onAddDashboards: (dashboard: SanitizedDashboardAsset[]) => Promise; + linkedDashboards: SanitizedDashboardAsset[]; + onClose: () => void; +}) { + const { + dependencies: { + start: { + streams: { streamsRepositoryClient }, + savedObjectsTagging: { ui: savedObjectsTaggingUi }, + }, + }, + } = useKibana(); + + const [query, setQuery] = useState(''); + + const [submittedQuery, setSubmittedQuery] = useState(query); + const [selectedDashboards, setSelectedDashboards] = useState([]); + const [isLoading, setIsLoading] = useState(false); + const [selectedTags, setSelectedTags] = useState([]); + const [isPopoverOpen, setIsPopoverOpen] = useState(false); + + const setSubmittedQueryDebounced = useMemo(() => { + return debounce(setSubmittedQuery, 150); + }, []); + + const dashboardSuggestionsFetch = useStreamsAppFetch( + ({ signal }) => { + return streamsRepositoryClient + .fetch('POST /api/streams/{id}/dashboards/_suggestions', { + signal, + params: { + path: { + id: entityId, + }, + query: { + query: submittedQuery, + }, + body: { + tags: selectedTags, + }, + }, + }) + .then(({ suggestions }) => { + return { + dashboards: suggestions.filter((dashboard) => { + return !linkedDashboards.find( + (linkedDashboard) => linkedDashboard.id === dashboard.id + ); + }), + }; + }); + }, + [streamsRepositoryClient, entityId, submittedQuery, selectedTags, linkedDashboards] + ); + + const tagList = savedObjectsTaggingUi.getTagList(); + + const button = ( + setIsPopoverOpen(!isPopoverOpen)} + isSelected={isPopoverOpen} + numFilters={tagList.length} + hasActiveFilters={selectedTags.length > 0} + numActiveFilters={selectedTags.length} + > + {i18n.translate('xpack.streams.addDashboardFlyout.filterButtonLabel', { + defaultMessage: 'Tags', + })} + + ); + + const filterGroupPopoverId = useGeneratedHtmlId({ + prefix: 'filterGroupPopover', + }); + + useEffect(() => { + setSelectedDashboards([]); + }, [linkedDashboards]); + + const allDashboards = useMemo(() => { + return dashboardSuggestionsFetch.value?.dashboards || []; + }, [dashboardSuggestionsFetch.value]); + + return ( + + + +

+ {i18n.translate('xpack.streams.addDashboardFlyout.flyoutHeaderLabel', { + defaultMessage: 'Add dashboards', + })} +

+ + + + + + {i18n.translate('xpack.streams.addDashboardFlyout.helpLabel', { + defaultMessage: + 'Select dashboards which you want to add and assign to the {stream} stream', + values: { + stream: entityId, + }, + })} + + + + { + setQuery(queryText); + setSubmittedQueryDebounced(queryText); + }} + /> + + + + setIsPopoverOpen(false)} + panelPaddingSize="none" + > + ({ + label: tag.name, + checked: selectedTags.includes(tag.id) ? 'on' : undefined, + }))} + onChange={(newOptions) => { + setSelectedTags( + newOptions + .filter((option) => option.checked === 'on') + .map((option) => savedObjectsTaggingUi.getTagIdFromName(option.label)!) + ); + }} + > + {(list, search) => ( +
+ {search} + {list} +
+ )} + + + + + + + + + + { + setIsLoading(true); + try { + await onAddDashboards(selectedDashboards); + } finally { + setIsLoading(false); + } + }} + > + {i18n.translate('xpack.streams.addDashboardFlyout.addDashboardsButtonLabel', { + defaultMessage: 'Add dashboards', + })} + + + + ); +} diff --git a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_dashboards_view/dashboard_table.tsx b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_dashboards_view/dashboard_table.tsx new file mode 100644 index 0000000000000..eb04553ad88b1 --- /dev/null +++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_dashboards_view/dashboard_table.tsx @@ -0,0 +1,85 @@ +/* + * 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 { EuiBasicTable, EuiBasicTableColumn, EuiFlexGroup, EuiFlexItem } from '@elastic/eui'; +import { i18n } from '@kbn/i18n'; +import React, { useMemo } from 'react'; +import type { SanitizedDashboardAsset } from '@kbn/streams-plugin/server/routes/dashboards/route'; +import { useKibana } from '../../hooks/use_kibana'; +import { tagListToReferenceList } from './to_reference_list'; + +export function DashboardsTable({ + dashboards, + compact = false, + selectedDashboards, + setSelectedDashboards, + loading, +}: { + loading: boolean; + dashboards: SanitizedDashboardAsset[] | undefined; + compact?: boolean; + selectedDashboards: SanitizedDashboardAsset[]; + setSelectedDashboards: (dashboards: SanitizedDashboardAsset[]) => void; +}) { + const { + dependencies: { + start: { + savedObjectsTagging: { ui: savedObjectsTaggingUi }, + }, + }, + } = useKibana(); + const columns = useMemo((): Array> => { + return [ + { + field: 'label', + name: i18n.translate('xpack.streams.dashboardTable.dashboardNameColumnTitle', { + defaultMessage: 'Dashboard name', + }), + }, + ...(!compact + ? ([ + { + field: 'tags', + name: i18n.translate('xpack.streams.dashboardTable.tagsColumnTitle', { + defaultMessage: 'Tags', + }), + render: (_, { tags }) => { + return ( + + + + ); + }, + }, + ] satisfies Array>) + : []), + ]; + }, [compact, savedObjectsTaggingUi]); + + const items = useMemo(() => { + return dashboards ?? []; + }, [dashboards]); + + return ( + + + { + setSelectedDashboards(newSelection); + }, + selected: selectedDashboards, + }} + /> + + ); +} diff --git a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_dashboards_view/index.tsx b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_dashboards_view/index.tsx new file mode 100644 index 0000000000000..fbb98b877e71a --- /dev/null +++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_dashboards_view/index.tsx @@ -0,0 +1,112 @@ +/* + * 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 { EuiButton, EuiFlexGroup, EuiFlexItem, EuiSearchBar } from '@elastic/eui'; +import { i18n } from '@kbn/i18n'; +import { StreamDefinition } from '@kbn/streams-schema'; +import React, { useMemo, useState } from 'react'; +import type { SanitizedDashboardAsset } from '@kbn/streams-plugin/server/routes/dashboards/route'; +import { AddDashboardFlyout } from './add_dashboard_flyout'; +import { DashboardsTable } from './dashboard_table'; +import { useDashboardsApi } from '../../hooks/use_dashboards_api'; +import { useDashboardsFetch } from '../../hooks/use_dashboards_fetch'; + +export function StreamDetailDashboardsView({ definition }: { definition?: StreamDefinition }) { + const [query, setQuery] = useState(''); + + const [isAddDashboardFlyoutOpen, setIsAddDashboardFlyoutOpen] = useState(false); + + const dashboardsFetch = useDashboardsFetch(definition?.name); + const { addDashboards, removeDashboards } = useDashboardsApi(definition?.name); + + const [isUnlinkLoading, setIsUnlinkLoading] = useState(false); + const linkedDashboards = useMemo(() => { + return dashboardsFetch.value?.dashboards ?? []; + }, [dashboardsFetch.value?.dashboards]); + + const filteredDashboards = useMemo(() => { + return linkedDashboards.filter((dashboard) => { + return dashboard.label.toLowerCase().includes(query.toLowerCase()); + }); + }, [linkedDashboards, query]); + + const [selectedDashboards, setSelectedDashboards] = useState([]); + + return ( + + + + {selectedDashboards.length > 0 && ( + { + try { + setIsUnlinkLoading(true); + + await removeDashboards(selectedDashboards); + await dashboardsFetch.refresh(); + + setSelectedDashboards([]); + } finally { + setIsUnlinkLoading(false); + } + }} + color="danger" + > + {i18n.translate('xpack.streams.streamDetailDashboardView.removeSelectedButtonLabel', { + defaultMessage: 'Unlink selected', + })} + + )} + { + setQuery(nextQuery.queryText); + }} + /> + { + setIsAddDashboardFlyoutOpen(true); + }} + > + {i18n.translate('xpack.streams.streamDetailDashboardView.addADashboardButtonLabel', { + defaultMessage: 'Add a dashboard', + })} + + + + + + {definition && isAddDashboardFlyoutOpen ? ( + { + await addDashboards(dashboards); + await dashboardsFetch.refresh(); + setIsAddDashboardFlyoutOpen(false); + }} + onClose={() => { + setIsAddDashboardFlyoutOpen(false); + }} + /> + ) : null} + + + ); +} diff --git a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_dashboards_view/to_reference_list.ts b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_dashboards_view/to_reference_list.ts new file mode 100644 index 0000000000000..7daa7d7660eb3 --- /dev/null +++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_dashboards_view/to_reference_list.ts @@ -0,0 +1,16 @@ +/* + * 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 { SavedObjectReference } from '@kbn/core/public'; + +export function tagListToReferenceList(tags: string[]): SavedObjectReference[] { + return tags.map((tag) => ({ + id: tag, + name: 'tag', + type: 'tag', + })); +} diff --git a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_view/index.tsx b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_view/index.tsx index 9ebc5a92f54db..148748bdc477e 100644 --- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_view/index.tsx +++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_view/index.tsx @@ -11,6 +11,7 @@ import { useStreamsAppParams } from '../../hooks/use_streams_app_params'; import { useStreamsAppFetch } from '../../hooks/use_streams_app_fetch'; import { useKibana } from '../../hooks/use_kibana'; import { StreamDetailOverview } from '../stream_detail_overview'; +import { StreamDetailDashboardsView } from '../stream_detail_dashboards_view'; import { StreamDetailManagement } from '../stream_detail_management'; export function StreamDetailView() { @@ -60,6 +61,13 @@ export function StreamDetailView() { defaultMessage: 'Overview', }), }, + { + name: 'dashboards', + content: , + label: i18n.translate('xpack.streams.streamDetailView.dashboardsTab', { + defaultMessage: 'Dashboards', + }), + }, { name: 'management', content: ( diff --git a/x-pack/solutions/observability/plugins/streams_app/public/hooks/use_dashboards_api.ts b/x-pack/solutions/observability/plugins/streams_app/public/hooks/use_dashboards_api.ts new file mode 100644 index 0000000000000..57f40a36cef68 --- /dev/null +++ b/x-pack/solutions/observability/plugins/streams_app/public/hooks/use_dashboards_api.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 { useCallback } from 'react'; +import { useAbortController } from '@kbn/observability-utils-browser/hooks/use_abort_controller'; +import type { SanitizedDashboardAsset } from '@kbn/streams-plugin/server/routes/dashboards/route'; +import { useKibana } from './use_kibana'; + +export const useDashboardsApi = (id?: string) => { + const { signal } = useAbortController(); + const { + dependencies: { + start: { + streams: { streamsRepositoryClient }, + }, + }, + } = useKibana(); + + const addDashboards = useCallback( + async (dashboards: SanitizedDashboardAsset[]) => { + if (!id) { + return; + } + + await streamsRepositoryClient.fetch('POST /api/streams/{id}/dashboards/_bulk', { + signal, + params: { + path: { + id, + }, + body: { + operations: dashboards.map((dashboard) => { + return { index: { id: dashboard.id } }; + }), + }, + }, + }); + }, + [id, signal, streamsRepositoryClient] + ); + + const removeDashboards = useCallback( + async (dashboards: SanitizedDashboardAsset[]) => { + if (!id) { + return; + } + await streamsRepositoryClient.fetch('POST /api/streams/{id}/dashboards/_bulk', { + signal, + params: { + path: { + id, + }, + body: { + operations: dashboards.map((dashboard) => { + return { delete: { id: dashboard.id } }; + }), + }, + }, + }); + }, + [id, signal, streamsRepositoryClient] + ); + + return { + addDashboards, + removeDashboards, + }; +}; diff --git a/x-pack/solutions/observability/plugins/streams_app/public/hooks/use_dashboards_fetch.ts b/x-pack/solutions/observability/plugins/streams_app/public/hooks/use_dashboards_fetch.ts new file mode 100644 index 0000000000000..c03e0cb2ea46f --- /dev/null +++ b/x-pack/solutions/observability/plugins/streams_app/public/hooks/use_dashboards_fetch.ts @@ -0,0 +1,37 @@ +/* + * 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 { useKibana } from './use_kibana'; +import { useStreamsAppFetch } from './use_streams_app_fetch'; + +export const useDashboardsFetch = (id?: string) => { + const { + dependencies: { + start: { + streams: { streamsRepositoryClient }, + }, + }, + } = useKibana(); + + const dashboardsFetch = useStreamsAppFetch( + ({ signal }) => { + if (!id) { + return Promise.resolve(undefined); + } + return streamsRepositoryClient.fetch('GET /api/streams/{id}/dashboards', { + signal, + params: { + path: { + id, + }, + }, + }); + }, + [id, streamsRepositoryClient] + ); + + return dashboardsFetch; +}; diff --git a/x-pack/solutions/observability/plugins/streams_app/public/hooks/use_streams_app_fetch.ts b/x-pack/solutions/observability/plugins/streams_app/public/hooks/use_streams_app_fetch.ts index 45911cbda851a..c70f5316a5847 100644 --- a/x-pack/solutions/observability/plugins/streams_app/public/hooks/use_streams_app_fetch.ts +++ b/x-pack/solutions/observability/plugins/streams_app/public/hooks/use_streams_app_fetch.ts @@ -11,6 +11,7 @@ import { useAbortableAsync, } from '@kbn/observability-utils-browser/hooks/use_abortable_async'; import { omit } from 'lodash'; +import { isRequestAbortedError } from '@kbn/server-route-repository-client'; import { useKibana } from './use_kibana'; export const useStreamsAppFetch: UseAbortableAsync<{}, { disableToastOnError?: boolean }> = ( @@ -25,7 +26,7 @@ export const useStreamsAppFetch: UseAbortableAsync<{}, { disableToastOnError?: b const onError = (error: Error) => { let requestUrl: string | undefined; - if (!options?.disableToastOnError) { + if (!options?.disableToastOnError && !isRequestAbortedError(error)) { if ( 'body' in error && typeof error.body === 'object' && diff --git a/x-pack/solutions/observability/plugins/streams_app/public/types.ts b/x-pack/solutions/observability/plugins/streams_app/public/types.ts index 680dd008d2e1f..8896a7aedfb4d 100644 --- a/x-pack/solutions/observability/plugins/streams_app/public/types.ts +++ b/x-pack/solutions/observability/plugins/streams_app/public/types.ts @@ -16,6 +16,7 @@ import type { import type { StreamsPluginSetup, StreamsPluginStart } from '@kbn/streams-plugin/public'; import type { UnifiedSearchPublicPluginStart } from '@kbn/unified-search-plugin/public'; import type { SharePublicSetup, SharePublicStart } from '@kbn/share-plugin/public/plugin'; +import type { SavedObjectTaggingPluginStart } from '@kbn/saved-objects-tagging-plugin/public'; import { NavigationPublicStart } from '@kbn/navigation-plugin/public/types'; /* eslint-disable @typescript-eslint/no-empty-interface*/ @@ -37,6 +38,7 @@ export interface StreamsAppStartDependencies { observabilityShared: ObservabilitySharedPluginStart; unifiedSearch: UnifiedSearchPublicPluginStart; share: SharePublicStart; + savedObjectsTagging: SavedObjectTaggingPluginStart; navigation: NavigationPublicStart; } diff --git a/x-pack/solutions/observability/plugins/streams_app/tsconfig.json b/x-pack/solutions/observability/plugins/streams_app/tsconfig.json index 7824c84d6ea6b..38923832b4b04 100644 --- a/x-pack/solutions/observability/plugins/streams_app/tsconfig.json +++ b/x-pack/solutions/observability/plugins/streams_app/tsconfig.json @@ -4,7 +4,6 @@ "outDir": "target/types" }, "include": [ - "../../../../../typings/**/*", "common/**/*", "public/**/*", "typings/**/*", @@ -14,18 +13,28 @@ ], "exclude": ["target/**/*", ".storybook/**/*.js"], "kbn_references": [ + "@kbn/i18n", + "@kbn/streams-plugin", "@kbn/core", "@kbn/data-plugin", "@kbn/data-views-plugin", "@kbn/observability-shared-plugin", "@kbn/unified-search-plugin", - "@kbn/react-kibana-context-render", + "@kbn/share-plugin", + "@kbn/navigation-plugin", + "@kbn/saved-objects-tagging-plugin", "@kbn/shared-ux-link-redirect-app", "@kbn/typed-react-router-config", - "@kbn/i18n", + "@kbn/react-kibana-context-render", + "@kbn/code-editor", "@kbn/observability-utils-browser", + "@kbn/observability-utils-server", + "@kbn/ui-theme", + "@kbn/calculate-auto", + "@kbn/core-notifications-browser", "@kbn/kibana-react-plugin", "@kbn/es-query", + "@kbn/server-route-repository-client", "@kbn/logging", "@kbn/deeplinks-observability", "@kbn/config-schema", diff --git a/x-pack/test/api_integration/apis/streams/assets/dashboard.ts b/x-pack/test/api_integration/apis/streams/assets/dashboard.ts new file mode 100644 index 0000000000000..7886a52da6165 --- /dev/null +++ b/x-pack/test/api_integration/apis/streams/assets/dashboard.ts @@ -0,0 +1,341 @@ +/* + * 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 expect from '@kbn/expect'; +import { enableStreams, indexDocument } from '../helpers/requests'; +import { createStreamsRepositorySupertestClient } from '../helpers/repository_client'; +import { FtrProviderContext } from '../../../ftr_provider_context'; +import { cleanUpRootStream } from '../helpers/cleanup'; + +export default function ({ getService }: FtrProviderContext) { + const supertest = getService('supertest'); + const esClient = getService('es'); + + const kibanaServer = getService('kibanaServer'); + + const apiClient = createStreamsRepositorySupertestClient(supertest); + + const SPACE_ID = 'default'; + const ARCHIVES = [ + 'test/api_integration/fixtures/kbn_archiver/saved_objects/search.json', + 'test/api_integration/fixtures/kbn_archiver/saved_objects/basic.json', + ]; + + const SEARCH_DASHBOARD_ID = 'b70c7ae0-3224-11e8-a572-ffca06da1357'; + const BASIC_DASHBOARD_ID = 'be3733a0-9efe-11e7-acb3-3dab96693fab'; + const BASIC_DASHBOARD_TITLE = 'Requests'; + + async function loadDashboards() { + for (const archive of ARCHIVES) { + await kibanaServer.importExport.load(archive, { space: SPACE_ID }); + } + } + + async function unloadDashboards() { + for (const archive of ARCHIVES) { + await kibanaServer.importExport.unload(archive, { space: SPACE_ID }); + } + } + + async function linkDashboard(id: string) { + const response = await apiClient.fetch('PUT /api/streams/{id}/dashboards/{dashboardId}', { + params: { path: { id: 'logs', dashboardId: id } }, + }); + + expect(response.status).to.be(200); + } + + async function unlinkDashboard(id: string) { + const response = await apiClient.fetch('DELETE /api/streams/{id}/dashboards/{dashboardId}', { + params: { path: { id: 'logs', dashboardId: id } }, + }); + + expect(response.status).to.be(200); + } + + async function bulkLinkDashboard(...ids: string[]) { + const response = await apiClient.fetch('POST /api/streams/{id}/dashboards/_bulk', { + params: { + path: { id: 'logs' }, + body: { + operations: ids.map((id) => { + return { + index: { + id, + }, + }; + }), + }, + }, + }); + + expect(response.status).to.be(200); + } + + async function bulkUnlinkDashboard(...ids: string[]) { + const response = await apiClient.fetch('POST /api/streams/{id}/dashboards/_bulk', { + params: { + path: { id: 'logs' }, + body: { + operations: ids.map((id) => { + return { + delete: { + id, + }, + }; + }), + }, + }, + }); + + expect(response.status).to.be(200); + } + + async function deleteAssetIndices() { + const concreteIndices = await esClient.indices.resolveIndex({ + name: '.kibana_streams_assets*', + }); + + if (concreteIndices.indices.length) { + await esClient.indices.delete({ + index: concreteIndices.indices.map((index) => index.name), + }); + } + } + + describe('Asset links', () => { + before(async () => { + await enableStreams(supertest); + + await indexDocument(esClient, 'logs', { + '@timestamp': '2024-01-01T00:00:10.000Z', + message: '2023-01-01T00:00:10.000Z error test', + }); + }); + + after(async () => { + await cleanUpRootStream(esClient); + + await esClient.indices.deleteDataStream({ + name: ['logs*'], + }); + + await deleteAssetIndices(); + }); + + describe('without writing', () => { + it('creates no indices initially', async () => { + const exists = await esClient.indices.exists({ index: '.kibana_streams_assets' }); + + expect(exists).to.eql(false); + }); + + it('creates no indices after reading the assets', async () => { + const response = await apiClient.fetch('GET /api/streams/{id}/dashboards', { + params: { path: { id: 'logs' } }, + }); + + expect(response.status).to.be(200); + + const exists = await esClient.indices.exists({ index: '.kibana_streams_assets' }); + + expect(exists).to.eql(false); + }); + }); + + describe('after linking a dashboard', () => { + before(async () => { + await loadDashboards(); + + await linkDashboard(SEARCH_DASHBOARD_ID); + }); + + after(async () => { + await unloadDashboards(); + await unlinkDashboard(SEARCH_DASHBOARD_ID); + }); + + it('creates the index', async () => { + const exists = await esClient.indices.exists({ index: '.kibana_streams_assets' }); + + expect(exists).to.be(true); + }); + + it('lists the dashboard in the stream response', async () => { + const response = await apiClient.fetch('GET /api/streams/{id}', { + params: { path: { id: 'logs' } }, + }); + + expect(response.status).to.eql(200); + + expect(response.body.dashboards?.length).to.eql(1); + }); + + it('lists the dashboard in the dashboards get response', async () => { + const response = await apiClient.fetch('GET /api/streams/{id}/dashboards', { + params: { path: { id: 'logs' } }, + }); + + expect(response.status).to.eql(200); + + expect(response.body.dashboards.length).to.eql(1); + }); + + describe('after manually rolling over the index and relinking the dashboard', () => { + before(async () => { + await esClient.indices.updateAliases({ + actions: [ + { + add: { + index: `.kibana_streams_assets-000001`, + alias: `.kibana_streams_assets`, + is_write_index: false, + }, + }, + ], + }); + + await esClient.indices.create({ + index: `.kibana_streams_assets-000002`, + }); + + await unlinkDashboard(SEARCH_DASHBOARD_ID); + await linkDashboard(SEARCH_DASHBOARD_ID); + }); + + it('there are no duplicates', async () => { + const response = await apiClient.fetch('GET /api/streams/{id}/dashboards', { + params: { path: { id: 'logs' } }, + }); + + expect(response.status).to.eql(200); + + expect(response.body.dashboards.length).to.eql(1); + + const esResponse = await esClient.search({ + index: `.kibana_streams_assets`, + }); + + expect(esResponse.hits.hits.length).to.eql(1); + }); + }); + + describe('after deleting the indices and relinking the dashboard', () => { + before(async () => { + await deleteAssetIndices(); + + await unlinkDashboard(SEARCH_DASHBOARD_ID); + await linkDashboard(SEARCH_DASHBOARD_ID); + }); + + it('recovers on write and lists the linked dashboard ', async () => { + const response = await apiClient.fetch('GET /api/streams/{id}/dashboards', { + params: { path: { id: 'logs' } }, + }); + + expect(response.status).to.eql(200); + + expect(response.body.dashboards.length).to.eql(1); + }); + }); + + describe('after deleting the dashboards', () => { + before(async () => { + await unloadDashboards(); + }); + + it('no longer lists the dashboard as a linked asset', async () => { + const response = await apiClient.fetch('GET /api/streams/{id}/dashboards', { + params: { path: { id: 'logs' } }, + }); + + expect(response.status).to.eql(200); + + expect(response.body.dashboards.length).to.eql(0); + }); + }); + }); + + describe('after using the bulk API', () => { + before(async () => { + await loadDashboards(); + + await bulkLinkDashboard(SEARCH_DASHBOARD_ID, BASIC_DASHBOARD_ID); + }); + + after(async () => { + await bulkUnlinkDashboard(SEARCH_DASHBOARD_ID, BASIC_DASHBOARD_ID); + await unloadDashboards(); + }); + + it('shows the linked dashboards', async () => { + const response = await apiClient.fetch('GET /api/streams/{id}/dashboards', { + params: { path: { id: 'logs' } }, + }); + + expect(response.body.dashboards.length).to.eql(2); + }); + + describe('after unlinking one dashboard', () => { + before(async () => { + await bulkUnlinkDashboard(SEARCH_DASHBOARD_ID); + }); + + it('only shows the remaining linked dashboard', async () => { + const response = await apiClient.fetch('GET /api/streams/{id}/dashboards', { + params: { path: { id: 'logs' } }, + }); + + expect(response.body.dashboards.length).to.eql(1); + + expect(response.body.dashboards[0].id).to.eql(BASIC_DASHBOARD_ID); + }); + }); + }); + + describe('suggestions', () => { + before(async () => { + await loadDashboards(); + + await linkDashboard(SEARCH_DASHBOARD_ID); + }); + + after(async () => { + await unlinkDashboard(SEARCH_DASHBOARD_ID); + await unloadDashboards(); + }); + + describe('after creating multiple dashboards', () => { + it('suggests dashboards to link', async () => { + const response = await apiClient.fetch('POST /api/streams/{id}/dashboards/_suggestions', { + params: { path: { id: 'logs' }, body: { tags: [] }, query: { query: '' } }, + }); + + expect(response.status).to.eql(200); + expect(response.body.suggestions.length).to.eql(2); + }); + + // TODO: needs a dataset with dashboards with tags + it.skip('filters suggested dashboards based on tags', () => {}); + + it('filters suggested dashboards based on the query', async () => { + const response = await apiClient.fetch('POST /api/streams/{id}/dashboards/_suggestions', { + params: { + path: { id: 'logs' }, + body: { tags: [] }, + query: { query: BASIC_DASHBOARD_TITLE }, + }, + }); + + expect(response.status).to.eql(200); + expect(response.body.suggestions.length).to.eql(1); + + expect(response.body.suggestions[0].id).to.eql(BASIC_DASHBOARD_ID); + }); + }); + }); + }); +} diff --git a/x-pack/test/api_integration/apis/streams/classic.ts b/x-pack/test/api_integration/apis/streams/classic.ts index 67d72bcb0a0ac..af23eb783ead5 100644 --- a/x-pack/test/api_integration/apis/streams/classic.ts +++ b/x-pack/test/api_integration/apis/streams/classic.ts @@ -6,19 +6,11 @@ */ import expect from '@kbn/expect'; -import { JsonObject } from '@kbn/utility-types'; -import { - deleteStream, - enableStreams, - fetchDocument, - getStream, - indexDocument, - listStreams, - putStream, -} from './helpers/requests'; -import { FtrProviderContext } from '../../ftr_provider_context'; import { waitForDocumentInIndex } from '../../../alerting_api_integration/observability/helpers/alerting_wait_for_helpers'; +import { FtrProviderContext } from '../../ftr_provider_context'; import { cleanUpRootStream } from './helpers/cleanup'; +import { createStreamsRepositorySupertestClient } from './helpers/repository_client'; +import { fetchDocument, indexDocument } from './helpers/requests'; export default function ({ getService }: FtrProviderContext) { const supertest = getService('supertest'); @@ -26,27 +18,40 @@ export default function ({ getService }: FtrProviderContext) { const retryService = getService('retry'); const logger = getService('log'); + const TEST_STREAM_NAME = 'logs-test-default'; + + const apiClient = createStreamsRepositorySupertestClient(supertest); + describe('Classic streams', () => { - after(async () => { - await cleanUpRootStream(esClient); + before(async () => { + await apiClient.fetch('POST /api/streams/_enable'); }); - before(async () => { - await enableStreams(supertest); + after(async () => { + await cleanUpRootStream(esClient); + await esClient.indices.deleteDataStream({ + name: ['logs*'], + }); }); it('Shows non-wired data streams', async () => { const doc = { message: '2023-01-01T00:00:10.000Z error test', }; - const response = await indexDocument(esClient, 'logs-test-default', doc); + const response = await indexDocument(esClient, TEST_STREAM_NAME, doc); expect(response.result).to.eql('created'); - const streams = await listStreams(supertest); - const classicStream = streams.streams.find( - (stream: JsonObject) => stream.name === 'logs-test-default' - ); + + const { + body: { streams }, + status, + } = await apiClient.fetch('GET /api/streams'); + + expect(status).to.eql(200); + + const classicStream = streams.find((stream) => stream.name === TEST_STREAM_NAME); + expect(classicStream).to.eql({ - name: 'logs-test-default', + name: TEST_STREAM_NAME, stream: { ingest: { processing: [], @@ -57,28 +62,44 @@ export default function ({ getService }: FtrProviderContext) { }); it('Allows setting processing on classic streams', async () => { - const response = await putStream(supertest, 'logs-test-default', { - ingest: { - processing: [ - { - config: { - grok: { - field: 'message', - patterns: [ - '%{TIMESTAMP_ISO8601:inner_timestamp} %{LOGLEVEL:log.level} %{GREEDYDATA:message2}', - ], + const putResponse = await apiClient.fetch('PUT /api/streams/{id}', { + params: { + path: { + id: TEST_STREAM_NAME, + }, + body: { + ingest: { + processing: [ + { + config: { + grok: { + field: 'message', + patterns: [ + '%{TIMESTAMP_ISO8601:inner_timestamp} %{LOGLEVEL:log.level} %{GREEDYDATA:message2}', + ], + }, + }, }, - }, + ], }, - ], - routing: [], + }, }, }); - expect(response).to.have.property('acknowledged', true); - const streamBody = await getStream(supertest, 'logs-test-default'); - expect(streamBody).to.eql({ - name: 'logs-test-default', - inherited_fields: {}, + + expect(putResponse.status).to.eql(200); + + expect(putResponse.body).to.have.property('acknowledged', true); + + const getResponse = await apiClient.fetch('GET /api/streams/{id}', { + params: { path: { id: TEST_STREAM_NAME } }, + }); + + expect(getResponse.status).to.eql(200); + + expect(getResponse.body).to.eql({ + name: TEST_STREAM_NAME, + dashboards: [], + inherited_fields: [], stream: { ingest: { processing: [ @@ -104,16 +125,16 @@ export default function ({ getService }: FtrProviderContext) { '@timestamp': '2024-01-01T00:00:10.000Z', message: '2023-01-01T00:00:10.000Z error test', }; - const response = await indexDocument(esClient, 'logs-test-default', doc); + const response = await indexDocument(esClient, TEST_STREAM_NAME, doc); expect(response.result).to.eql('created'); await waitForDocumentInIndex({ esClient, - indexName: 'logs-test-default', + indexName: TEST_STREAM_NAME, retryService, logger, docCountTarget: 2, }); - const result = await fetchDocument(esClient, 'logs-test-default', response._id); + const result = await fetchDocument(esClient, TEST_STREAM_NAME, response._id); expect(result._source).to.eql({ '@timestamp': '2024-01-01T00:00:10.000Z', message: '2023-01-01T00:00:10.000Z error test', @@ -126,13 +147,21 @@ export default function ({ getService }: FtrProviderContext) { }); it('Allows removing processing on classic streams', async () => { - const response = await putStream(supertest, 'logs-test-default', { - ingest: { - processing: [], - routing: [], + const response = await apiClient.fetch('PUT /api/streams/{id}', { + params: { + path: { id: TEST_STREAM_NAME }, + body: { + ingest: { + processing: [], + routing: [], + }, + }, }, }); - expect(response).to.have.property('acknowledged', true); + + expect(response.status).to.eql(200); + + expect(response.body).to.have.property('acknowledged', true); }); it('Executes processing on classic streams after removing processing', async () => { @@ -140,16 +169,16 @@ export default function ({ getService }: FtrProviderContext) { // default logs pipeline fills in timestamp with current date if not set message: '2023-01-01T00:00:10.000Z info mylogger this is the message', }; - const response = await indexDocument(esClient, 'logs-test-default', doc); + const response = await indexDocument(esClient, TEST_STREAM_NAME, doc); expect(response.result).to.eql('created'); await waitForDocumentInIndex({ esClient, - indexName: 'logs-test-default', + indexName: TEST_STREAM_NAME, retryService, logger, docCountTarget: 3, }); - const result = await fetchDocument(esClient, 'logs-test-default', response._id); + const result = await fetchDocument(esClient, TEST_STREAM_NAME, response._id); expect(result._source).to.eql({ // accept any date '@timestamp': (result._source as { [key: string]: unknown })['@timestamp'], @@ -158,10 +187,22 @@ export default function ({ getService }: FtrProviderContext) { }); it('Allows deleting classic streams', async () => { - await deleteStream(supertest, 'logs-test-default'); - const streams = await listStreams(supertest); - const classicStream = streams.streams.find( - (stream: JsonObject) => stream.name === 'logs-test-default' + const deleteStreamResponse = await apiClient.fetch('DELETE /api/streams/{id}', { + params: { + path: { + id: TEST_STREAM_NAME, + }, + }, + }); + + expect(deleteStreamResponse.status).to.eql(200); + + const getStreamsResponse = await apiClient.fetch('GET /api/streams'); + + expect(getStreamsResponse.status).to.eql(200); + + const classicStream = getStreamsResponse.body.streams.find( + (stream) => stream.name === TEST_STREAM_NAME ); expect(classicStream).to.eql(undefined); }); diff --git a/x-pack/test/api_integration/apis/streams/config.ts b/x-pack/test/api_integration/apis/streams/config.ts index c737db9499836..2fbcac9ff3b8d 100644 --- a/x-pack/test/api_integration/apis/streams/config.ts +++ b/x-pack/test/api_integration/apis/streams/config.ts @@ -5,12 +5,26 @@ * 2.0. */ -import { FtrConfigProviderContext } from '@kbn/test'; +import { FtrConfigProviderContext, getKibanaCliLoggers } from '@kbn/test'; export default async function ({ readConfigFile }: FtrConfigProviderContext) { const baseIntegrationTestsConfig = await readConfigFile(require.resolve('../../config.ts')); return { ...baseIntegrationTestsConfig.getAll(), + kbnTestServer: { + ...baseIntegrationTestsConfig.get('kbnTestServer'), + serverArgs: [ + ...baseIntegrationTestsConfig.get('kbnTestServer.serverArgs'), + `--logging.loggers=${JSON.stringify([ + ...getKibanaCliLoggers(baseIntegrationTestsConfig.get('kbnTestServer.serverArgs')), + { + name: 'plugins.streams', + level: 'debug', + appenders: ['default'], + }, + ])}`, + ], + }, testFiles: [require.resolve('.')], }; } diff --git a/x-pack/test/api_integration/apis/streams/enrichment.ts b/x-pack/test/api_integration/apis/streams/enrichment.ts index e9fb604438ee6..01c67ac14808e 100644 --- a/x-pack/test/api_integration/apis/streams/enrichment.ts +++ b/x-pack/test/api_integration/apis/streams/enrichment.ts @@ -20,14 +20,17 @@ export default function ({ getService }: FtrProviderContext) { const logger = getService('log'); describe('Enrichment', () => { - after(async () => { - await cleanUpRootStream(esClient); - }); - before(async () => { await enableStreams(supertest); }); + after(async () => { + await cleanUpRootStream(esClient); + await esClient.indices.deleteDataStream({ + name: ['logs*'], + }); + }); + it('Place processing steps', async () => { const body: WiredStreamConfigDefinition = { ingest: { diff --git a/x-pack/test/api_integration/apis/streams/flush_config.ts b/x-pack/test/api_integration/apis/streams/flush_config.ts index b04b5ff7959a9..d8b95ef92e62b 100644 --- a/x-pack/test/api_integration/apis/streams/flush_config.ts +++ b/x-pack/test/api_integration/apis/streams/flush_config.ts @@ -6,99 +6,101 @@ */ import expect from '@kbn/expect'; -import { SearchTotalHits } from '@elastic/elasticsearch/lib/api/typesWithBodyKey'; -import { StreamDefinition } from '@kbn/streams-schema'; -import { deleteStream, enableStreams, indexDocument } from './helpers/requests'; +import { ClientRequestParamsOf } from '@kbn/server-route-repository-utils'; +import type { StreamsRouteRepository } from '@kbn/streams-plugin/server'; import { FtrProviderContext } from '../../ftr_provider_context'; -import { waitForDocumentInIndex } from '../../../alerting_api_integration/observability/helpers/alerting_wait_for_helpers'; import { cleanUpRootStream } from './helpers/cleanup'; +import { createStreamsRepositorySupertestClient } from './helpers/repository_client'; +import { enableStreams, indexDocument } from './helpers/requests'; -const streams: StreamDefinition[] = [ +type StreamPutItem = ClientRequestParamsOf< + StreamsRouteRepository, + 'PUT /api/streams/{id}' +>['params']['body'] & { name: string }; + +const streams: StreamPutItem[] = [ { name: 'logs', - stream: { - ingest: { - processing: [], - wired: { - fields: { - '@timestamp': { - type: 'date', - }, - message: { - type: 'match_only_text', - }, - 'host.name': { - type: 'keyword', - }, - 'log.level': { - type: 'keyword', - }, + ingest: { + processing: [], + wired: { + fields: { + '@timestamp': { + type: 'date', }, - }, - routing: [ - { - name: 'logs.test', - condition: { - and: [ - { - field: 'numberfield', - operator: 'gt', - value: 15, - }, - ], - }, + message: { + type: 'match_only_text', }, - { - name: 'logs.test2', - condition: { - and: [ - { - field: 'field2', - operator: 'eq', - value: 'abc', - }, - ], - }, + 'host.name': { + type: 'keyword', + }, + 'log.level': { + type: 'keyword', }, - ], + }, }, + routing: [ + { + name: 'logs.test', + condition: { + and: [ + { + field: 'numberfield', + operator: 'gt', + value: 15, + }, + ], + }, + }, + { + name: 'logs.test2', + condition: { + and: [ + { + field: 'field2', + operator: 'eq', + value: 'abc', + }, + ], + }, + }, + ], }, }, { name: 'logs.test', - stream: { - ingest: { - processing: [], - wired: { - fields: {}, + ingest: { + processing: [], + wired: { + fields: { + numberfield: { + type: 'long', + }, }, - routing: [], }, }, }, { name: 'logs.test2', - stream: { - ingest: { - processing: [ - { - config: { - grok: { - field: 'message', - patterns: ['%{NUMBER:numberfield}'], - }, + ingest: { + processing: [ + { + config: { + grok: { + field: 'message', + patterns: ['%{NUMBER:numberfield}'], }, }, - ], - wired: { - fields: { - numberfield: { - type: 'long', - }, + }, + ], + wired: { + fields: { + field2: { + type: 'keyword', }, }, - routing: [], }, + routing: [], }, }, ]; @@ -106,34 +108,68 @@ const streams: StreamDefinition[] = [ export default function ({ getService }: FtrProviderContext) { const supertest = getService('supertest'); const esClient = getService('es'); - const retryService = getService('retry'); - const logger = getService('log'); + + const apiClient = createStreamsRepositorySupertestClient(supertest); // An anticipated use case is that a user will want to flush a tree of streams from a config file describe('Flush from config file', () => { after(async () => { - await deleteStream(supertest, 'logs.nginx'); await cleanUpRootStream(esClient); + await esClient.indices.deleteDataStream({ + name: ['logs*'], + }); }); - // Note: Each step is dependent on the previous - it('Enable streams', async () => { + before(async () => { await enableStreams(supertest); + await createStreams(); + await indexDocuments(); }); - it('PUTs all streams one by one without errors', async () => { - for (const { name, stream } of streams) { - const response = await supertest - .put(`/api/streams/${name}`) - .set('kbn-xsrf', 'xxx') - .send(stream) - .expect(200); + it('puts the data in the right data streams', async () => { + const logsResponse = await esClient.search({ + index: 'logs', + query: { + match: { 'log.level': 'info' }, + }, + }); - expect(response.body).to.have.property('acknowledged', true); - } + expect(logsResponse.hits.total).to.eql({ value: 1, relation: 'eq' }); + + const logsTestResponse = await esClient.search({ + index: 'logs.test', + query: { + match: { numberfield: 20 }, + }, + }); + + expect(logsTestResponse.hits.total).to.eql({ value: 1, relation: 'eq' }); + + const logsTest2Response = await esClient.search({ + index: 'logs.test2', + query: { + match: { field2: 'abc' }, + }, + }); + + expect(logsTest2Response.hits.total).to.eql({ value: 1, relation: 'eq' }); }); - it('send data and it is handled properly', async () => { + async function createStreams() { + for (const { name: streamId, ...stream } of streams) { + await apiClient + .fetch('PUT /api/streams/{id}', { + params: { + body: stream, + path: { id: streamId }, + }, + }) + .expect(200) + .then((response) => expect(response.body.acknowledged).to.eql(true)); + } + } + + async function indexDocuments() { // send data that stays in logs const doc = { '@timestamp': '2024-01-01T00:00:00.000Z', @@ -142,7 +178,6 @@ export default function ({ getService }: FtrProviderContext) { }; const response = await indexDocument(esClient, 'logs', doc); expect(response.result).to.eql('created'); - await waitForDocumentInIndex({ esClient, indexName: 'logs', retryService, logger }); // send data that lands in logs.test const doc2 = { @@ -152,7 +187,6 @@ export default function ({ getService }: FtrProviderContext) { }; const response2 = await indexDocument(esClient, 'logs', doc2); expect(response2.result).to.eql('created'); - await waitForDocumentInIndex({ esClient, indexName: 'logs.test', retryService, logger }); // send data that lands in logs.test2 const doc3 = { @@ -162,15 +196,6 @@ export default function ({ getService }: FtrProviderContext) { }; const response3 = await indexDocument(esClient, 'logs', doc3); expect(response3.result).to.eql('created'); - await waitForDocumentInIndex({ esClient, indexName: 'logs.test2', retryService, logger }); - }); - - it('makes data searchable as expected', async () => { - const query = { - match: { numberfield: 123 }, - }; - const response = await esClient.search({ index: 'logs.test2', query }); - expect((response.hits.total as SearchTotalHits).value).to.eql(1); - }); + } }); } diff --git a/x-pack/test/api_integration/apis/streams/full_flow.ts b/x-pack/test/api_integration/apis/streams/full_flow.ts index fd46df8002d74..610f160f7419b 100644 --- a/x-pack/test/api_integration/apis/streams/full_flow.ts +++ b/x-pack/test/api_integration/apis/streams/full_flow.ts @@ -6,13 +6,7 @@ */ import expect from '@kbn/expect'; -import { - deleteStream, - enableStreams, - fetchDocument, - forkStream, - indexDocument, -} from './helpers/requests'; +import { enableStreams, fetchDocument, forkStream, indexDocument } from './helpers/requests'; import { FtrProviderContext } from '../../ftr_provider_context'; import { waitForDocumentInIndex } from '../../../alerting_api_integration/observability/helpers/alerting_wait_for_helpers'; import { cleanUpRootStream } from './helpers/cleanup'; @@ -25,8 +19,10 @@ export default function ({ getService }: FtrProviderContext) { describe('Basic functionality', () => { after(async () => { - await deleteStream(supertest, 'logs.nginx'); await cleanUpRootStream(esClient); + await esClient.indices.deleteDataStream({ + name: ['logs*'], + }); }); // Note: Each step is dependent on the previous diff --git a/x-pack/test/api_integration/apis/streams/helpers/repository_client.ts b/x-pack/test/api_integration/apis/streams/helpers/repository_client.ts new file mode 100644 index 0000000000000..581243abbba11 --- /dev/null +++ b/x-pack/test/api_integration/apis/streams/helpers/repository_client.ts @@ -0,0 +1,18 @@ +/* + * 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 { StreamsRouteRepository } from '@kbn/streams-plugin/server'; +import supertest from 'supertest'; +import { + RepositorySupertestClient, + getApiClientFromSupertest, +} from '../../../../common/utils/server_route_repository/create_supertest_service_from_repository'; + +export function createStreamsRepositorySupertestClient( + st: supertest.Agent +): RepositorySupertestClient { + return getApiClientFromSupertest(st); +} diff --git a/x-pack/test/api_integration/apis/streams/helpers/requests.ts b/x-pack/test/api_integration/apis/streams/helpers/requests.ts index f26c53aaaf0aa..799ec480c4a02 100644 --- a/x-pack/test/api_integration/apis/streams/helpers/requests.ts +++ b/x-pack/test/api_integration/apis/streams/helpers/requests.ts @@ -18,7 +18,7 @@ export async function enableStreams(supertest: Agent) { } export async function indexDocument(esClient: Client, index: string, document: JsonObject) { - const response = await esClient.index({ index, document }); + const response = await esClient.index({ index, document, refresh: 'wait_for' }); return response; } @@ -38,13 +38,13 @@ export async function forkStream(supertest: Agent, root: string, body: JsonObjec } export async function putStream(supertest: Agent, name: string, body: StreamConfigDefinition) { - const req = supertest.put(`/api/streams/${name}`).set('kbn-xsrf', 'xxx'); + const req = supertest.put(`/api/streams/${encodeURIComponent(name)}`).set('kbn-xsrf', 'xxx'); const response = await req.send(body).expect(200); return response.body; } export async function getStream(supertest: Agent, name: string) { - const req = supertest.get(`/api/streams/${name}`).set('kbn-xsrf', 'xxx'); + const req = supertest.get(`/api/streams/${encodeURIComponent(name)}`).set('kbn-xsrf', 'xxx'); const response = await req.send().expect(200); return response.body; } diff --git a/x-pack/test/api_integration/apis/streams/index.ts b/x-pack/test/api_integration/apis/streams/index.ts index 6c4cf358b8ac3..299f349172ffd 100644 --- a/x-pack/test/api_integration/apis/streams/index.ts +++ b/x-pack/test/api_integration/apis/streams/index.ts @@ -13,6 +13,7 @@ export default function ({ loadTestFile }: FtrProviderContext) { loadTestFile(require.resolve('./enrichment')); loadTestFile(require.resolve('./classic')); loadTestFile(require.resolve('./flush_config')); + loadTestFile(require.resolve('./assets/dashboard')); loadTestFile(require.resolve('./schema')); }); } diff --git a/x-pack/test/common/utils/server_route_repository/create_supertest_service_from_repository.ts b/x-pack/test/common/utils/server_route_repository/create_supertest_service_from_repository.ts new file mode 100644 index 0000000000000..4c0b3da777a1e --- /dev/null +++ b/x-pack/test/common/utils/server_route_repository/create_supertest_service_from_repository.ts @@ -0,0 +1,211 @@ +/* + * 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 { + formatRequest, + ServerRouteRepository, + EndpointOf, + ReturnOf, + ClientRequestParamsOf, +} from '@kbn/server-route-repository'; +import supertest from 'supertest'; +import { Subtract, RequiredKeys } from 'utility-types'; +import { format, UrlObject } from 'url'; +import { kbnTestConfig } from '@kbn/test'; + +type MaybeOptional> = RequiredKeys extends never + ? [TArgs] | [] + : [TArgs]; + +export interface RepositorySupertestClient { + fetch: >( + endpoint: TEndpoint, + ...options: MaybeOptional< + { + type?: 'form-data'; + } & ClientRequestParamsOf + > + ) => RepositorySupertestReturnOf; +} + +type RepositorySupertestReturnOf< + TServerRouteRepository extends ServerRouteRepository, + TEndpoint extends EndpointOf +> = OverwriteThisMethods< + WithoutPromise, + Promise<{ + text: string; + status: number; + body: ReturnOf; + }> +>; + +type ScopedApiClientWithBasicAuthFactory = ( + kibanaServer: UrlObject, + username: string +) => RepositorySupertestClient; + +type ApiClientFromSupertestFactory = ( + st: supertest.Agent +) => RepositorySupertestClient; + +interface RepositorySupertestClientFactory { + getScopedApiClientWithBasicAuth: ScopedApiClientWithBasicAuthFactory; + getApiClientFromSupertest: ApiClientFromSupertestFactory; +} + +export function createSupertestClientFactoryFromRepository< + TServerRouteRepository extends ServerRouteRepository +>(): RepositorySupertestClientFactory { + return { + getScopedApiClientWithBasicAuth: (kibanaServer, username) => { + return getScopedApiClientWithBasicAuth(kibanaServer, username); + }, + getApiClientFromSupertest: (st) => { + return getApiClientFromSupertest(st); + }, + }; +} + +function getScopedApiClientWithBasicAuth( + kibanaServer: UrlObject, + username: string +): RepositorySupertestClient { + const { password } = kbnTestConfig.getUrlParts(); + const baseUrlWithAuth = format({ + ...kibanaServer, + auth: `${username}:${password}`, + }); + + return getApiClientFromSupertest(supertest(baseUrlWithAuth)); +} + +export function getApiClientFromSupertest( + st: supertest.Agent +): RepositorySupertestClient { + return { + fetch: (endpoint, ...rest) => { + const options = rest.length ? rest[0] : { type: undefined }; + + const { type } = options; + + const params = 'params' in options ? (options.params as Record) : {}; + + const { method, pathname, version } = formatRequest(endpoint, params.path); + const url = format({ pathname, query: params?.query }); + + const headers: Record = { 'kbn-xsrf': 'foo' }; + + if (version) { + headers['Elastic-Api-Version'] = version; + } + + let res: supertest.Test; + if (type === 'form-data') { + const fields: Array<[string, any]> = Object.entries(params.body); + const formDataRequest = st[method](url) + .set(headers) + .set('Content-type', 'multipart/form-data'); + + for (const field of fields) { + void formDataRequest.field(field[0], field[1]); + } + + res = formDataRequest; + } else if (params.body) { + res = st[method](url).send(params.body).set(headers); + } else { + res = st[method](url).set(headers); + } + + return res as RepositorySupertestReturnOf< + TServerRouteRepository, + EndpointOf + >; + }, + }; +} + +type WithoutPromise> = Subtract>; + +// this is a little intense, but without it, method overrides are lost +// e.g., { +// end(one:string) +// end(one:string, two:string) +// } +// would lose the first signature. This keeps up to eight signatures. +type OverloadedParameters = T extends { + (...args: infer A1): any; + (...args: infer A2): any; + (...args: infer A3): any; + (...args: infer A4): any; + (...args: infer A5): any; + (...args: infer A6): any; + (...args: infer A7): any; + (...args: infer A8): any; +} + ? A1 | A2 | A3 | A4 | A5 | A6 | A7 | A8 + : T extends { + (...args: infer A1): any; + (...args: infer A2): any; + (...args: infer A3): any; + (...args: infer A4): any; + (...args: infer A5): any; + (...args: infer A6): any; + (...args: infer A7): any; + } + ? A1 | A2 | A3 | A4 | A5 | A6 | A7 + : T extends { + (...args: infer A1): any; + (...args: infer A2): any; + (...args: infer A3): any; + (...args: infer A4): any; + (...args: infer A5): any; + (...args: infer A6): any; + } + ? A1 | A2 | A3 | A4 | A5 | A6 + : T extends { + (...args: infer A1): any; + (...args: infer A2): any; + (...args: infer A3): any; + (...args: infer A4): any; + (...args: infer A5): any; + } + ? A1 | A2 | A3 | A4 | A5 + : T extends { + (...args: infer A1): any; + (...args: infer A2): any; + (...args: infer A3): any; + (...args: infer A4): any; + } + ? A1 | A2 | A3 | A4 + : T extends { + (...args: infer A1): any; + (...args: infer A2): any; + (...args: infer A3): any; + } + ? A1 | A2 | A3 + : T extends { + (...args: infer A1): any; + (...args: infer A2): any; + } + ? A1 | A2 + : T extends (...args: infer A) => any + ? A + : any; + +type OverrideReturnType any, TNextReturnType> = ( + ...args: OverloadedParameters +) => WithoutPromise> & TNextReturnType; + +type OverwriteThisMethods, TNextReturnType> = TNextReturnType & { + [key in keyof T]: T[key] extends (...args: infer TArgs) => infer TReturnType + ? TReturnType extends Promise + ? OverrideReturnType + : (...args: TArgs) => TReturnType + : T[key]; +}; diff --git a/x-pack/test/tsconfig.json b/x-pack/test/tsconfig.json index a591c2b6b4573..c44db155280c4 100644 --- a/x-pack/test/tsconfig.json +++ b/x-pack/test/tsconfig.json @@ -190,6 +190,8 @@ "@kbn/integration-assistant-plugin", "@kbn/core-elasticsearch-server", "@kbn/streams-schema", + "@kbn/server-route-repository-utils", + "@kbn/streams-plugin", "@kbn/response-ops-rule-params" ] }