[Dataset Quality] Apply chunking algorithm for getIndexBasicStats #221153

rStelmach · 2025-05-21T16:31:17Z

After adding a chunking algorithm for getIndexBasicStats errors are not appearing anymore.
For testing purposes I created 600 indicies using this command :
for i in {1..600}; do curl -u <username>:<password> -X POST "<elasticsearch_url>/<index_base_name>-$i/_doc" -H 'Content-Type: application/json' -d '{"@timestamp": "<timestamp>", "message": "<message>"}' done

and created this test for testing the telemetry function :


import { ElasticsearchClient } from '@kbn/core-elasticsearch-server';
import { Client } from '@elastic/elasticsearch';
import { lastValueFrom } from 'rxjs';
import {
  getIndexBasicStats,
  addMappingsToIndices,
  getIndicesInfoForPattern,
} from '@kbn/dataset-quality-plugin/server/services/data_telemetry/helpers';

export type IndicesGetResponse = Record<IndexName, any>;
export type IndexName = string;

async function runTest() {
  const client = new Client({
    node: 'http://localhost:9200',
    auth: { username: '', password: '' },
  });

  const results = {
    addMappings: await testAddMappingsToIndicesTypeSafety(client),
    indicesInfo: await testGetIndicesInfoForPattern(client),
    basicStats: await testGetIndexBasicStats(client),
  };

  process.stdout.write('Test results:\n');
  process.stdout.write(`- addMappingsToIndices: ${results.addMappings ? 'SUCCESS' : 'FAILED'}\n`);
  process.stdout.write(
    `- getIndicesInfoForPattern: ${results.indicesInfo ? 'SUCCESS' : 'FAILED'}\n`
  );
  process.stdout.write(`- getIndexBasicStats: ${results.basicStats ? 'SUCCESS' : 'FAILED'}\n`);
}

async function testAddMappingsToIndicesTypeSafety(client: Client): Promise<boolean> {
  try {
    const logsIndexPatterns = [{ patternName: 'logs-*', shipper: 'test', pattern: 'logs-*' }];
    const dataStreamsInfo = [
      {
        patternName: 'logs-*',
        shipper: 'test',
        isDataStream: false,
        name: 'logs-test-1',
        indices: ['logs-test-1'],
        mapping: undefined,
      },
    ];
    const observable = addMappingsToIndices({
      esClient: client as unknown as ElasticsearchClient,
      dataStreamsInfo,
      logsIndexPatterns,
    });
    const result = await lastValueFrom(observable);

    const hasMapping = result.length > 0 && !!result[0].mapping;
    process.stdout.write(
      `addMappingsToIndices ${hasMapping ? 'found mappings' : 'no mapping found'}\n`
    );
    if (hasMapping) {
      const keys = Object.keys(result[0].mapping!);
      process.stdout.write(`  Keys: ${keys.join(', ')}\n`);
    }
    return true;
  } catch (error) {
    process.stdout.write(
      `addMappingsToIndices error: ${error instanceof Error ? error.message : String(error)}\n`
    );
    return false;
  }
}

async function testGetIndicesInfoForPattern(client: Client): Promise<boolean> {
  try {
    const pattern = { patternName: 'logs-*', shipper: 'test', pattern: 'logs-*' };
    const result = await getIndicesInfoForPattern({
      esClient: client as unknown as ElasticsearchClient,
      pattern,
    });

    if (result.length > 0) {
      process.stdout.write(
        `getIndicesInfoForPattern found ${result.length} indices, sample: ${result[0].name}\n`
      );
      return true;
    } else {
      process.stdout.write('getIndicesInfoForPattern returned no results\n');
      return false;
    }
  } catch (error) {
    process.stdout.write(
      `getIndicesInfoForPattern error: ${error instanceof Error ? error.message : String(error)}\n`
    );
    return false;
  }
}

async function testGetIndexBasicStats(client: Client): Promise<boolean> {
  try {
    const response = await client.cat.indices({
      format: 'json',
      h: 'index',
      index: 'logs-*',
    });

    const indicesList = response as Array<{ index: string }>;

    if (indicesList.length === 0) {
      process.stdout.write('No indices found to test with\n');
      return false;
    }

    const indexBasicInfos = indicesList.map((indexObj) => ({
      patternName: 'logs-*',
      shipper: 'test',
      isDataStream: false,
      name: indexObj.index,
      indices: [indexObj.index],
      mapping: undefined,
    }));

    process.stdout.write(`Testing with ${indexBasicInfos.length} indices\n`);

    const observable = getIndexBasicStats({
      esClient: client as unknown as ElasticsearchClient,
      indices: indexBasicInfos,
      breatheDelay: 0,
    });

    const result = await lastValueFrom(observable);
    process.stdout.write(`getIndexBasicStats processed ${result.length} indices\n`);
    return result.length > 0;
  } catch (error) {
    process.stdout.write(
      `getIndexBasicStats error: ${error instanceof Error ? error.message : String(error)}\n`
    );
    return false;
  }
}

runTest().catch((err) => {
  process.stdout.write('Unhandled error in test:', err);
});

mohamedhamed-ahmed · 2025-05-22T13:01:18Z

@rStelmach I think there might be a couple of others that we are missing.

I was having a look at these truncations problems and came across a comment from @weltenwort
here. it summarises the issue and mentioned also another function with a similar problem getIndicesInfoForPattern .

Would be great if we can also add some error handling there and 1 suggestion is that we might also have a look at using features or filter_path (don't remember which) prob on the esClinet api call. So that we reduce the response to only the info that we need
ex:

await esClient.indices.get({
      index: pattern.pattern,
      features: ['mappings'],
    });

…attern

rStelmach · 2025-05-23T15:37:39Z

@mohamedhamed-ahmed
I updated few more functions and added some small error handling( I couldn't think of a better one). Let me know what you think

…tasks-generates-too-big-requests-and-responses

mohamedhamed-ahmed · 2025-05-26T16:19:53Z

@mohamedhamed-ahmed I updated few more functions and added some small error handling( I couldn't think of a better one). Let me know what you think

Thanks for applying the changes to the other functions. I see what you are trying to do there but would prefer if @awahab07 would have a look at the changes to make sure that the fallbacks are what we expect there and if the approach is fine from a telemetry prespective.

awahab07

Great to see the refactor to optimize ES calls and handling the output for errors.

Left some comments as certain error handlings may overfit and suppress some errors, which may result in data inconsistencies.

About where to focus what optimization and how to move forward, I suggest:

addMappingsToIndices could potentially result in errors where the output is too large and less likely to cause too_long_http_line_exception. So filtering the output is crucial in this function (chunking is still good here though).
getIndexBasicStats looks like the main culprit causing too_long_http_line_exception errors. So request chunking, filtering output and handling large output errors is important here.
To ensure the applied filtering is stripping required fields, it'd be great to have a scenario to test the Before and After effect to make sure the resultant stats are intact. For this, a mix of data streams and regular indices with names conforming to log patterns can be created, while collected fields being ingested. Also a few documents in the failure store for a few log patterns present.

awahab07 · 2025-05-28T03:44:08Z