Skip to content

Commit dfca496

Browse files
github-actions[bot]github-actions[bot]
committed
Implement saved query substitution for S3 integrations (#1705)
* Add example queries for cloudfront Signed-off-by: Simeon Widdis <[email protected]> * Add Flint S3 label to haproxy Signed-off-by: Simeon Widdis <[email protected]> * Add query substitution method to build process Signed-off-by: Simeon Widdis <[email protected]> * Fix tests Signed-off-by: Simeon Widdis <[email protected]> * Update build API with new data source fields Signed-off-by: Simeon Widdis <[email protected]> * Update frontend to use new API term Signed-off-by: Simeon Widdis <[email protected]> * Use new fields in original integration creation request Signed-off-by: Simeon Widdis <[email protected]> * Fix broken data source usage ref Signed-off-by: Simeon Widdis <[email protected]> --------- Signed-off-by: Simeon Widdis <[email protected]> (cherry picked from commit 0b58e64) Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
1 parent 780e1e8 commit dfca496

File tree

12 files changed

+142
-38
lines changed

12 files changed

+142
-38
lines changed

‎docs/integrations/setup.md

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,8 @@ the author in hindsight.
2727
If working on S3-based integrations, it's worth noting that queries have some values
2828
[substituted](https://github.com/opensearch-project/dashboards-observability/blob/4e1e0e585/public/components/integrations/components/setup_integration.tsx#L438) when installing. They are:
2929

30+
- `{table_name}` is the fully qualified name of the Flint table, typically `datasource.database.object_name`.
31+
This is also substituted in any linked Saved Queries when using S3-based integrations.
3032
- `{s3_bucket_location}` to locate data.
3133
- `{s3_checkpoint_location}` to store intermediate results, which is required by Spark.
3234
- `{object_name}` used for giving tables a unique name per-integration to avoid collisions.

‎public/components/integrations/components/create_integration_helpers.ts

Lines changed: 27 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -223,12 +223,14 @@ const createIndexMapping = async (
223223
});
224224
};
225225

226-
const createDataSourceMappings = async (
226+
const createIndexPatternMappings = async (
227227
targetDataSource: string,
228228
integrationTemplateId: string,
229229
integration: IntegrationConfig,
230230
setToast: (title: string, color?: Color, text?: string | undefined) => void
231231
): Promise<void> => {
232+
// TODO the nested methods still need the dataSource -> indexPattern rename applied, sub-methods
233+
// here still have old naming convention
232234
const http = coreRefs.http!;
233235
const data = await http.get(`${INTEGRATIONS_BASE}/repository/${integrationTemplateId}/schema`);
234236
let error: string | null = null;
@@ -282,25 +284,42 @@ export async function addIntegrationRequest(
282284
integration: IntegrationConfig,
283285
setToast: (title: string, color?: Color, text?: string | undefined) => void,
284286
name?: string,
285-
dataSource?: string,
287+
indexPattern?: string,
286288
workflows?: string[],
287-
skipRedirect?: boolean
289+
skipRedirect?: boolean,
290+
dataSourceInfo?: { dataSource: string; tableName: string }
288291
): Promise<boolean> {
289292
const http = coreRefs.http!;
290293
if (addSample) {
291-
createDataSourceMappings(
294+
createIndexPatternMappings(
292295
`ss4o_${integration.type}-${integrationTemplateId}-*-sample`,
293296
integrationTemplateId,
294297
integration,
295298
setToast
296299
);
297300
name = `${integrationTemplateId}-sample`;
298-
dataSource = `ss4o_${integration.type}-${integrationTemplateId}-sample-sample`;
301+
indexPattern = `ss4o_${integration.type}-${integrationTemplateId}-sample-sample`;
302+
}
303+
304+
const createReqBody: {
305+
name?: string;
306+
indexPattern?: string;
307+
workflows?: string[];
308+
dataSource?: string;
309+
tableName?: string;
310+
} = {
311+
name,
312+
indexPattern,
313+
workflows,
314+
};
315+
if (dataSourceInfo) {
316+
createReqBody.dataSource = dataSourceInfo.dataSource;
317+
createReqBody.tableName = dataSourceInfo.tableName;
299318
}
300319

301320
let response: boolean = await http
302321
.post(`${INTEGRATIONS_BASE}/store/${templateName}`, {
303-
body: JSON.stringify({ name, dataSource, workflows }),
322+
body: JSON.stringify(createReqBody),
304323
})
305324
.then((res) => {
306325
setToast(`${name} integration successfully added!`, 'success');
@@ -326,13 +345,13 @@ export async function addIntegrationRequest(
326345
});
327346
const requestBody =
328347
data.sampleData
329-
.map((record) => `{"create": { "_index": "${dataSource}" } }\n${JSON.stringify(record)}`)
348+
.map((record) => `{"create": { "_index": "${indexPattern}" } }\n${JSON.stringify(record)}`)
330349
.join('\n') + '\n';
331350
response = await http
332351
.post(CONSOLE_PROXY, {
333352
body: requestBody,
334353
query: {
335-
path: `${dataSource}/_bulk?refresh=wait_for`,
354+
path: `${indexPattern}/_bulk?refresh=wait_for`,
336355
method: 'POST',
337356
},
338357
})

‎public/components/integrations/components/setup_integration.tsx

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -435,11 +435,12 @@ export function SetupIntegrationFormInputs({
435435
);
436436
}
437437

438+
const makeTableName = (config: IntegrationSetupInputs): string => {
439+
return `${config.connectionDataSource}.default.${config.connectionTableName}`;
440+
};
441+
438442
const prepareQuery = (query: string, config: IntegrationSetupInputs): string => {
439-
let queryStr = query.replaceAll(
440-
'{table_name}',
441-
`${config.connectionDataSource}.default.${config.connectionTableName}`
442-
);
443+
let queryStr = query.replaceAll('{table_name}', makeTableName(config));
443444
queryStr = queryStr.replaceAll('{s3_bucket_location}', config.connectionLocation);
444445
queryStr = queryStr.replaceAll('{s3_checkpoint_location}', config.checkpointLocation);
445446
queryStr = queryStr.replaceAll('{object_name}', config.connectionTableName);
@@ -516,7 +517,8 @@ const addIntegration = async ({
516517
config.displayName,
517518
`flint_${config.connectionDataSource}_default_${config.connectionTableName}__*`,
518519
config.enabledWorkflows,
519-
setIsInstalling ? true : false
520+
setIsInstalling ? true : false,
521+
{ dataSource: config.connectionDataSource, tableName: makeTableName(config) }
520522
);
521523
if (setIsInstalling) {
522524
setIsInstalling(false, res);

‎server/adaptors/integrations/__data__/repository/aws_cloudfront/assets/example_queries-1.0.0.ndjson

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
{"attributes":{"createdTimeMs":1713289099101,"savedQuery":{"data_sources":"[{\"name\":\"mys3\",\"type\":\"s3glue\",\"label\":\"mys3\",\"value\":\"mys3\"}]","description":"","name":"Top IPs by Request Count","query":"SELECT c_ip, COUNT(*) AS request_count FROM {table_name} GROUP BY c_ip ORDER BY request_count DESC LIMIT 10;","query_lang":"SQL","selected_date_range":{"end":"now","start":"now-15m","text":""},"selected_fields":{"text":"","tokens":[]},"selected_timestamp":{"name":"","type":"timestamp"}},"title":"Top IPs by Request Count","version":1},"id":"1d07d010-fc18-11ee-99c9-43e5dbd0692c","references":[],"type":"observability-search","updated_at":"2024-04-16T17:52:30.414Z","version":"WzI3NTEsMV0="}
2+
{"attributes":{"createdTimeMs":1713293044079,"savedQuery":{"data_sources":"[{\"name\":\"mys3\",\"type\":\"s3glue\",\"label\":\"mys3\",\"value\":\"mys3\"}]","description":"","name":"Top Status by Count","query":"SELECT sc_status, COUNT(*) AS status_count FROM {table_name} GROUP BY sc_status ORDER BY status_count DESC LIMIT 10;","query_lang":"SQL","selected_date_range":{"end":"now","start":"now-15m","text":""},"selected_fields":{"text":"","tokens":[]},"selected_timestamp":{"name":"","type":"timestamp"}},"title":"Top Status by Count","version":1},"id":"4c6b8820-fc21-11ee-ab45-d3075d0510e6","references":[],"type":"observability-search","updated_at":"2024-04-16T18:44:47.956Z","version":"WzI4MzAsMV0="}
3+
{"attributes":{"createdTimeMs":1713290175184,"savedQuery":{"data_sources":"[{\"name\":\"mys3\",\"type\":\"s3glue\",\"label\":\"mys3\",\"value\":\"mys3\"}]","description":"","name":"Number of Requests","query":"SELECT COUNT(*) AS request_count FROM {table_name};","query_lang":"SQL","selected_date_range":{"end":"now","start":"now-15m","text":""},"selected_fields":{"text":"","tokens":[]},"selected_timestamp":{"name":"","type":"timestamp"}},"title":"Number of Requests","version":1},"id":"9e6a9b40-fc1a-11ee-99c9-43e5dbd0692c","references":[],"type":"observability-search","updated_at":"2024-04-16T17:56:15.220Z","version":"WzI3NTIsMV0="}
4+
{"attributes":{"createdTimeMs":1713293161193,"savedQuery":{"data_sources":"[{\"name\":\"mys3\",\"type\":\"s3glue\",\"label\":\"mys3\",\"value\":\"mys3\"}]","description":"","name":"Total Bytes Served","query":"SELECT SUM(sc_bytes) AS total_bytes_served FROM {table_name};","query_lang":"SQL","selected_date_range":{"end":"now","start":"now-15m","text":""},"selected_fields":{"text":"","tokens":[]},"selected_timestamp":{"name":"","type":"timestamp"}},"title":"Total Bytes Served","version":1},"id":"92398eb0-fc21-11ee-ab45-d3075d0510e6","references":[],"type":"observability-search","updated_at":"2024-04-16T18:46:01.242Z","version":"WzI4MzEsMV0="}
5+
{"attributes":{"createdTimeMs":1713293269224,"savedQuery":{"data_sources":"[{\"name\":\"mys3\",\"type\":\"s3glue\",\"label\":\"mys3\",\"value\":\"mys3\"}]","description":"","name":"Average Time Taken","query":"SELECT AVG(time_taken) AS average_time_taken FROM {table_name};","query_lang":"SQL","selected_date_range":{"end":"now","start":"now-15m","text":""},"selected_fields":{"text":"","tokens":[]},"selected_timestamp":{"name":"","type":"timestamp"}},"title":"Average Time Taken","version":1},"id":"d2a038a0-fc21-11ee-ab45-d3075d0510e6","references":[],"type":"observability-search","updated_at":"2024-04-16T18:47:49.290Z","version":"WzI4MzIsMV0="}
6+
{"attributes":{"createdTimeMs":1713293425335,"savedQuery":{"data_sources":"[{\"name\":\"mys3\",\"type\":\"s3glue\",\"label\":\"mys3\",\"value\":\"mys3\"}]","description":"","name":"Slow Requests from Average Time threshold","query":"WITH avg_time AS (SELECT AVG(time_to_first_byte) AS avg_time FROM {table_name}) SELECT * FROM {table_name} CROSS JOIN avg_time WHERE time_to_first_byte > 1 * avg_time LIMIT 10;","query_lang":"SQL","selected_date_range":{"end":"now","start":"now-15m","text":""},"selected_fields":{"text":"","tokens":[]},"selected_timestamp":{"name":"","type":"timestamp"}},"title":"Slow Requests from Average Time threshold","version":1},"id":"2fac4250-fc22-11ee-ab45-d3075d0510e6","references":[],"type":"observability-search","updated_at":"2024-04-16T18:59:34.785Z","version":"WzI4MzQsMV0="}
7+
{"attributes":{"createdTimeMs":1713294061574,"savedQuery":{"data_sources":"[{\"name\":\"mys3\",\"type\":\"s3glue\",\"label\":\"mys3\",\"value\":\"mys3\"}]","description":"","name":"Requests by User Agent","query":"SELECT * FROM {table_name} WHERE cs_user_agent LIKE '%Chrome%' LIMIT 10;","query_lang":"SQL","selected_date_range":{"end":"now","start":"now-15m","text":""},"selected_fields":{"text":"","tokens":[]},"selected_timestamp":{"name":"","type":"timestamp"}},"title":"Requests by User Agent","version":1},"id":"aae73c80-fc23-11ee-ab45-d3075d0510e6","references":[],"type":"observability-search","updated_at":"2024-04-16T19:01:01.640Z","version":"WzI4MzUsMV0="}
8+
{"exportedCount":7,"missingRefCount":0,"missingReferences":[]}

‎server/adaptors/integrations/__data__/repository/aws_cloudfront/aws_cloudfront-1.0.0.json

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,13 @@
7272
"extension": "sql",
7373
"type": "query",
7474
"workflows": ["dashboards"]
75+
},
76+
{
77+
"name": "example_queries",
78+
"version": "1.0.0",
79+
"extension": "ndjson",
80+
"type": "savedObjectBundle",
81+
"workflows": ["queries"]
7582
}
7683
],
7784
"sampleData": {

‎server/adaptors/integrations/__data__/repository/haproxy/haproxy-1.0.0.json

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55
"description": "Analyze HAProxy access logs.",
66
"license": "Apache-2.0",
77
"type": "logs",
8-
"labels": ["Observability", "Logs"],
8+
"labels": ["Observability", "Logs", "Flint S3"],
99
"author": "OpenSearch",
1010
"sourceUrl": "https://github.com/opensearch-project/dashboards-observability/tree/main/server/adaptors/integrations/__data__/repository/haproxy/info",
1111
"workflows": [

‎server/adaptors/integrations/__test__/builder.test.ts

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,7 @@ describe('IntegrationInstanceBuilder', () => {
5858
describe('build', () => {
5959
it('should build an integration instance', async () => {
6060
const options = {
61-
dataSource: 'instance-datasource',
61+
indexPattern: 'instance-datasource',
6262
name: 'instance-name',
6363
};
6464

@@ -131,7 +131,7 @@ describe('IntegrationInstanceBuilder', () => {
131131

132132
it('should reject with an error if integration is not valid', async () => {
133133
const options = {
134-
dataSource: 'instance-datasource',
134+
indexPattern: 'instance-datasource',
135135
name: 'instance-name',
136136
};
137137
jest
@@ -143,7 +143,7 @@ describe('IntegrationInstanceBuilder', () => {
143143

144144
it('should reject with an error if getAssets rejects', async () => {
145145
const options = {
146-
dataSource: 'instance-datasource',
146+
indexPattern: 'instance-datasource',
147147
name: 'instance-name',
148148
};
149149

@@ -160,7 +160,7 @@ describe('IntegrationInstanceBuilder', () => {
160160

161161
it('should reject with an error if postAssets throws an error', async () => {
162162
const options = {
163-
dataSource: 'instance-datasource',
163+
indexPattern: 'instance-datasource',
164164
name: 'instance-name',
165165
};
166166
const remappedAssets = [
@@ -297,7 +297,7 @@ describe('IntegrationInstanceBuilder', () => {
297297
},
298298
];
299299
const options = {
300-
dataSource: 'instance-datasource',
300+
indexPattern: 'instance-datasource',
301301
name: 'instance-name',
302302
};
303303
const expectedInstance = {
@@ -333,7 +333,7 @@ describe('IntegrationInstanceBuilder', () => {
333333
},
334334
];
335335
const options = {
336-
dataSource: 'instance-datasource',
336+
indexPattern: 'instance-datasource',
337337
name: 'instance-name',
338338
};
339339

‎server/adaptors/integrations/__test__/manager.test.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -242,7 +242,7 @@ describe('IntegrationsKibanaBackend', () => {
242242
expect(mockRepository.getIntegration).toHaveBeenCalledWith(templateName);
243243
expect(instanceBuilder.build).toHaveBeenCalledWith(template, {
244244
name,
245-
dataSource: 'datasource',
245+
indexPattern: 'datasource',
246246
});
247247
expect(mockSavedObjectsClient.create).toHaveBeenCalledWith(
248248
'integration-instance',

‎server/adaptors/integrations/integrations_adaptor.ts

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,10 @@ export interface IntegrationsAdaptor {
1717
loadIntegrationInstance: (
1818
templateName: string,
1919
name: string,
20-
dataSource: string,
21-
workflows?: string[]
20+
indexPattern: string,
21+
workflows?: string[],
22+
dataSource?: string,
23+
tableName?: string
2224
) => Promise<IntegrationInstance>;
2325

2426
deleteIntegrationInstance: (id: string) => Promise<unknown>;

‎server/adaptors/integrations/integrations_builder.ts

Lines changed: 66 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -11,8 +11,10 @@ import { deepCheck } from './repository/utils';
1111

1212
interface BuilderOptions {
1313
name: string;
14-
dataSource: string;
14+
indexPattern: string;
1515
workflows?: string[];
16+
dataSource?: string;
17+
tableName?: string;
1618
}
1719

1820
interface SavedObject {
@@ -42,12 +44,67 @@ export class IntegrationInstanceBuilder {
4244
return Promise.reject(assets.error);
4345
}
4446
const remapped = this.remapIDs(this.getSavedObjectBundles(assets.value, options.workflows));
45-
const withDataSource = this.remapDataSource(remapped, options.dataSource);
46-
const refs = await this.postAssets(withDataSource);
47+
const withDataSource = this.remapDataSource(remapped, options.indexPattern);
48+
const withSubstitutedQueries = this.substituteQueries(
49+
withDataSource,
50+
options.dataSource,
51+
options.tableName
52+
);
53+
const refs = await this.postAssets(withSubstitutedQueries as SavedObjectsBulkCreateObject[]);
4754
const builtInstance = await this.buildInstance(integration, refs, options);
4855
return builtInstance;
4956
}
5057

58+
// If we have a data source or table specified, hunt for saved queries and update them with the
59+
// new DS/table.
60+
substituteQueries(assets: SavedObject[], dataSource?: string, tableName?: string): SavedObject[] {
61+
if (!dataSource) {
62+
return assets;
63+
}
64+
65+
assets = assets.map((asset) => {
66+
if (asset.type === 'observability-search') {
67+
const savedQuery = ((asset.attributes as unknown) as {
68+
savedQuery: {
69+
// The actual SavedSearchAttributes type uses "dataSources", but when exporting it's
70+
// "data_sources". I'm not sure why the discrepancy exists but since that's the exported
71+
// format we need to define our own type here.
72+
data_sources: string;
73+
query: string;
74+
query_lang: string;
75+
};
76+
}).savedQuery;
77+
if (!savedQuery.data_sources) {
78+
return asset;
79+
}
80+
const dataSources = JSON.parse(savedQuery.data_sources) as Array<{
81+
name: string;
82+
type: string;
83+
label: string;
84+
value: string;
85+
}>;
86+
for (const ds of dataSources) {
87+
if (ds.type !== 's3glue') {
88+
continue; // Nothing to do
89+
}
90+
// TODO is there a distinction between these where we should only set one? They're all
91+
// equivalent in every export I've seen.
92+
ds.name = dataSource;
93+
ds.label = dataSource;
94+
ds.value = dataSource;
95+
}
96+
savedQuery.data_sources = JSON.stringify(dataSources);
97+
98+
if (savedQuery.query_lang === 'SQL' && tableName) {
99+
savedQuery.query = savedQuery.query.replaceAll('{table_name}', tableName);
100+
}
101+
}
102+
return asset;
103+
});
104+
105+
return assets;
106+
}
107+
51108
getSavedObjectBundles(
52109
assets: ParsedIntegrationAsset[],
53110
includeWorkflows?: string[]
@@ -69,18 +126,14 @@ export class IntegrationInstanceBuilder {
69126
.flat() as SavedObject[];
70127
}
71128

72-
remapDataSource(
73-
assets: SavedObject[],
74-
dataSource: string | undefined
75-
): Array<{ type: string; attributes: { title: string } }> {
129+
remapDataSource(assets: SavedObject[], dataSource: string | undefined): SavedObject[] {
76130
if (!dataSource) return assets;
77-
assets = assets.map((asset) => {
131+
return assets.map((asset) => {
78132
if (asset.type === 'index-pattern') {
79133
asset.attributes.title = dataSource;
80134
}
81135
return asset;
82136
});
83-
return assets;
84137
}
85138

86139
remapIDs(assets: SavedObject[]): SavedObject[] {
@@ -136,7 +189,10 @@ export class IntegrationInstanceBuilder {
136189
return Promise.resolve({
137190
name: options.name,
138191
templateName: config.value.name,
139-
dataSource: options.dataSource,
192+
// Before data sources existed we called the index pattern a data source. Now we need the old
193+
// name for BWC but still use the new data sources in building, so we map the variable only
194+
// for returned output here
195+
dataSource: options.indexPattern,
140196
creationDate: new Date().toISOString(),
141197
assets: refs,
142198
});

‎server/adaptors/integrations/integrations_manager.ts

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -157,8 +157,10 @@ export class IntegrationsManager implements IntegrationsAdaptor {
157157
loadIntegrationInstance = async (
158158
templateName: string,
159159
name: string,
160-
dataSource: string,
161-
workflows?: string[]
160+
indexPattern: string,
161+
workflows?: string[],
162+
dataSource?: string,
163+
tableName?: string
162164
): Promise<IntegrationInstance> => {
163165
const template = await this.repository.getIntegration(templateName);
164166
if (template === null) {
@@ -171,8 +173,10 @@ export class IntegrationsManager implements IntegrationsAdaptor {
171173
addRequestToMetric('integrations', 'create', 'count');
172174
const result = await this.instanceBuilder.build(template, {
173175
name,
174-
dataSource,
176+
indexPattern,
175177
workflows,
178+
dataSource,
179+
tableName,
176180
});
177181
const test = await this.client.create('integration-instance', result);
178182
return Promise.resolve({ ...result, id: test.id });

‎server/routes/integrations/integrations_router.ts

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -81,8 +81,10 @@ export function registerIntegrationsRoute(router: IRouter) {
8181
}),
8282
body: schema.object({
8383
name: schema.string(),
84-
dataSource: schema.string(),
84+
indexPattern: schema.string(),
8585
workflows: schema.maybe(schema.arrayOf(schema.string())),
86+
dataSource: schema.maybe(schema.string()),
87+
tableName: schema.maybe(schema.string()),
8688
}),
8789
},
8890
},
@@ -92,8 +94,10 @@ export function registerIntegrationsRoute(router: IRouter) {
9294
return a.loadIntegrationInstance(
9395
request.params.templateName,
9496
request.body.name,
97+
request.body.indexPattern,
98+
request.body.workflows,
9599
request.body.dataSource,
96-
request.body.workflows
100+
request.body.tableName
97101
);
98102
});
99103
}

0 commit comments

Comments
 (0)