加载中

Kibana data.search 服务

可以通过多种方式搜索存储在 Elasticsearch 中的数据，例如使用 Elasticsearch REST API 或使用 Elasticsearch Client 进行低级访问。

然而，搜索 Elasticsearch 推荐且最简单的方法是使用低级搜索服务。该服务由 data 插件提供，使用它不仅可以访问存储的数据，还可以使用各种功能，例如自定义搜索策略 (Custom Search Strategies)、异步搜索 (Asynchronous Search)、部分结果 (Partial Results)、搜索会话 (Search Sessions) 等。

以下是从自定义插件中使用 data.search 服务的基本示例

import { CoreStart, Plugin } from '@kbn/core/public';
import { DataPublicPluginStart, isCompleteResponse, isErrorResponse } from '@kbn/data-plugin/public';

export interface MyPluginStartDependencies {
  data: DataPublicPluginStart;
}

export class MyPlugin implements Plugin {
  public start(core: CoreStart, { data }: MyPluginStartDependencies) {
    const query = {
      filter: [{
        match_all: {}
      }],
    };
    const req = {
      params: {
        index: 'my-index-*',
        body: {
          query,
          aggs: {},
        },
      }
    };
    data.search.search(req).subscribe({
      next: (result) => {
        if (isCompleteResponse(res)) {
          // handle search result
        } else if (isErrorResponse(res)) {
          // handle error, this means that some results were returned, but the search has failed to complete.
        } else {
          // handle partial results if you want.
        }
      },
      error: (e) => {
        // handle error thrown, for example a server hangup
      },
    })
  }
}

注意：data 插件包含有助于生成 query 和 aggs 部分的服务，以及使用 data.indexPatterns 服务管理索引的服务。

注意：服务器端搜索

data.search 服务在服务器端和客户端均可用，且 API 类似。

search 方法可能会抛出几种类型的错误，例如

  • EsError 用于源自 Elasticsearch 的错误
  • PainlessError 用于源自 Painless 脚本的错误
  • AbortError 如果搜索通过 AbortController 被中止
  • HttpError 如果发生网络错误

若要在应用程序上下文中显示错误，请使用 data.search 服务提供的辅助方法。这些错误会使用 core.notifications 服务以 Toast 消息的形式显示。

data.search.search(req).subscribe({
  next: (result) => {},
  error: (e) => {
    data.search.showError(e);
  },
});

如果您决定自行处理错误，请注意来自 Elasticsearch 的错误。它们有一个额外的 attributes 属性，其中包含来自 Elasticsearch 的原始错误信息。

data.search.search(req).subscribe({
  next: (result) => {},
  error: (e) => {
    if (e instanceof IEsError) {
      showErrorReason(e.attributes);
    }
  },
});

搜索服务的 search 方法支持名为 options 的第二个参数。其中一个选项提供了 abortSignal，如果不再需要结果，可以用它来停止正在运行的搜索。

import { AbortError } from '@kbn/kibana-utils-plugin/common';

const abortController = new AbortController();
data.search
  .search(req, {
    abortSignal: abortController.signal,
  })
  .subscribe({
    next: (result) => {
      // handle result
    },
    error: (e) => {
      if (e instanceof AbortError) {
        // you can ignore this error
        return;
      }
      // handle error, for example a server hangup
    },
  });

// Abort the search request after a second
setTimeout(() => {
  abortController.abort();
}, 1000);
重要：如果不再需要结果，请取消搜索

用户可能不再对搜索结果感兴趣。例如，他们可能会开始新的搜索或离开您的应用程序而不等待结果。您应该通过在搜索 API 中使用 AbortController 来处理此类情况。

默认情况下，搜索服务使用 DSL 查询和聚合语法，并按原样返回 Elasticsearch 的响应。它还提供了一些额外的基础策略，例如 Async DSL（x-pack 默认）和 EQL。

例如，要使用 data.search 服务运行 EQL 查询，您需要在 options 参数中指定策略名称

const req = getEqlRequest();
data.search
  .search(req, {
    strategy: EQL_SEARCH_STRATEGY,
  })
  .subscribe({
    next: (result) => {
      // handle EQL result
    },
  });

要使用不同的查询语法、预处理请求或在将响应返回给客户端之前处理它，您可以创建并注册自定义搜索策略来封装您的自定义逻辑。

以下示例展示了如何定义、注册和使用一种搜索策略，该策略在将请求发送到默认 DSL 搜索策略之前对请求进行预处理，然后在返回之前处理响应。

// ./myPlugin/server/myStrategy.ts

/**
 * Your custom search strategy should implement the ISearchStrategy interface, requiring at minimum a `search` function.
 */
export const mySearchStrategyProvider = (
  data: PluginStart
): ISearchStrategy<IMyStrategyRequest, IMyStrategyResponse> => {
  const preprocessRequest = (request: IMyStrategyRequest) => {
    // Custom preprocessing
  };

  const formatResponse = (response: IMyStrategyResponse) => {
    // Custom post-processing
  };

  // Get the default search strategy
  const es = data.search.getSearchStrategy(ES_SEARCH_STRATEGY);
  return {
    search: (request, options, deps) => {
      return formatResponse(es.search(preprocessRequest(request), options, deps));
    },
  };
};
		
// ./myPlugin/server/plugin.ts
import type { CoreSetup, CoreStart, Plugin } from '@kbn/core/server';

import { mySearchStrategyProvider } from './my_strategy';

/**
 * Your plugin will receive the `data` plugin contact in both the setup and start lifecycle hooks.
 */
export interface MyPluginSetupDeps {
  data: PluginSetup;
}

export interface MyPluginStartDeps {
  data: PluginStart;
}

/**
 * In your custom server side plugin, register the strategy from the setup contract
 */
export class MyPlugin implements Plugin {
  public setup(core: CoreSetup<MyPluginStartDeps>, deps: MyPluginSetupDeps) {
    core.getStartServices().then(([_, depsStart]) => {
      const myStrategy = mySearchStrategyProvider(depsStart.data);
      deps.data.search.registerSearchStrategy('myCustomStrategy', myStrategy);
    });
  }
}
		
// ./myPlugin/public/plugin.ts
const req = getRequest();
data.search
  .search(req, {
    strategy: 'myCustomStrategy',
  })
  .subscribe({
    next: (result) => {
      // handle result
    },
  });

开源的默认搜索策略 (ES_SEARCH_STRATEGY) 会同步运行搜索，在查询执行期间保持与 Elasticsearch 的开放连接。这些查询的持续时间受 kibana.yml 中 elasticsearch.requestTimeout 设置的限制，默认值为 30 秒。

这种同步执行在大多数情况下效果很好。然而，随着 数据层 (data tiers) 和 运行时字段 (runtime fields) 等功能的引入，对允许运行较慢的查询（保持开放连接可能效率低下）的需求有所增加。在 7.7 版本中，Elasticsearch 引入了 async_search API，允许查询运行更长时间而无需保持开放连接。相反，初始搜索请求会返回一个 ID，用于标识在 Elasticsearch 中运行的搜索。该 ID 随后可用于检索、取消或管理搜索结果。

async_search API 是驱动更多高级 Kibana search 功能的基础，例如 partial results 和 search sessions。当可用时，Kibana 的默认搜索策略会自动设置为 async 默认搜索策略 (ENHANCED_ES_SEARCH_STRATEGY)，使 Kibana 能够运行更长的查询，并具有由 UI 设置 search:timeout 定义的可选持续时间限制。

如果您要实现自己的异步自定义搜索策略，请确保实现 cancel 和 extend，如下例所示

// ./myPlugin/server/myEnhancedStrategy.ts
export const myEnhancedSearchStrategyProvider = (
  data: PluginStart
): ISearchStrategy<IMyStrategyRequest, IMyStrategyResponse> => {
  // Get the default search strategy
  const ese = data.search.getSearchStrategy(ENHANCED_ES_SEARCH_STRATEGY);
  return {
    search: (request, options, deps) => {
      // search will be called multiple times,
      // be sure your response formatting is capable of handling partial results, as well as the final result.
      return formatResponse(ese.search(request, options, deps));
    },
    cancel: async (id, options, deps) => {
      // call the cancel method of the async strategy you are using or implement your own cancellation function.
      await ese.cancel(id, options, deps);
    },
    extend: async (id, keepAlive, options, deps) => {
      // async search results are not stored indefinitely. By default, they expire after 7 days (or as defined by data.search.sessions.defaultExpiration setting in kibana.yml).
      // call the extend method of the async strategy you are using or implement your own extend function.
      await ese.extend(id, options, deps);
    },
  };
};

高级搜索服务是一种简化创建和运行搜索请求的方法，无需编写自定义 DSL 查询。

function searchWithSearchSource() {
  const indexPattern = data.indexPatterns.getDefault();
  const query = data.query.queryString.getQuery();
  const filters = data.query.filterManager.getFilters();
  const timefilter = data.query.timefilter.timefilter.createFilter(indexPattern);
  if (timefilter) {
    filters.push(timefilter);
  }

  const searchSource = await data.search.searchSource.create();

  searchSource
    .setField('index', indexPattern)
    .setField('filter', filters)
    .setField('query', query)
    .setField('fields', selectedFields.length ? selectedFields.map((f) => f.name) : ['*'])
    .setField('aggs', getAggsDsl());

  searchSource.fetch$().subscribe({
    next: () => {},
    error: () => {},
  });
}

当使用 async 策略（例如异步 DSL 和异步 EQL）进行搜索时，搜索服务将流式传输返回部分结果。

虽然您可以忽略部分结果并等待最终结果后再进行渲染，但您也可以使用部分结果来为用户创建更具交互性的体验。但是，强烈建议确保用户知晓他们看到的是部分结果。

// Handling partial results
data.search.search(req).subscribe({
  next: (result) => {
    if (isCompleteResponse(res)) {
      renderFinalResult(res);
    } else if (isPartialResponse(res)) {
      renderPartialResult(res);
    }
  },
});

// Skipping partial results
const finalResult = await data.search.search(req).toPromise();

搜索会话是一个比搜索更高级的概念。搜索会话描述了一个或多个带有附加上下文的异步搜索请求的分组。

当您希望允许用户异步运行某些操作（例如，长时间跨度的仪表板），并在稍后快速恢复结果时，搜索会话非常有用。搜索服务会透明地从 .async-search 索引中获取结果，而不是再次运行每个请求。

在内部，搜索会话内运行的任何搜索都会保存到一个对象中，从而允许 Kibana 管理其生命周期。大多数已保存对象会在短时间后自动删除，但如果用户选择保存搜索会话，则会持久化保存该对象，以便稍后可以恢复结果。

已存储的搜索会话列在管理 (Management) 应用程序中的 Kibana > 搜索会话 (Search Sessions) 下，可以轻松找到、管理和恢复它们。

作为开发人员，您可能会遇到这两个常见的用例

  • 在现有搜索会话中运行搜索
  • 在您的应用程序中支持搜索会话

对于此示例，假设您正在实现一种将在仪表板上显示的新型 Embeddable（可嵌入组件）。然而，同样的原则适用于您运行的任何搜索请求，只要您在其中运行的应用程序正在管理活动会话即可。

因为仪表板应用程序已经在管理搜索会话，您所要做的就是将 searchSessionId 参数传递给任何 search 调用。这适用于低级和高级搜索 API。

搜索信息将被添加到搜索会话的已保存对象中。

export class SearchEmbeddable extends Embeddable<MyInput, MyOutput> {
  private async fetchData() {
    // Every embeddable receives an optional `searchSessionId` input parameter.
    const { searchSessionId } = this.input;

    // Setup your search source
    this.configureSearchSource();

    try {
      // Mark the embeddable as loading
      this.updateOutput({ loading: true, error: undefined });

      // Make the request, wait for the final result
      const { rawResponse: resp } = await searchSource
        .fetch$({
          sessionId: searchSessionId,
        })
        .toPromise();

      this.useSearchResult(resp);

      this.updateOutput({ loading: false, error: undefined });
    } catch (error) {
      // handle search errors
      this.updateOutput({ loading: false, error });
    }
  }
}

您也可以直接从 搜索服务 中检索活动的 Search Session ID

async function fetchData(data: DataPublicPluginStart) {
  try {
    return await searchSource
      .fetch$({
        sessionId: data.search.sessions.getSessionId(),
      })
      .toPromise();
  } catch (e) {
    // handle search errors
  }
}
注意：使用服务器端搜索的搜索会话

搜索会话由客户端发起。如果您使用的路由运行服务器端搜索，您可以将 searchSessionId 发送到服务器，然后将其传递给服务器端的 data.search 函数调用。

在您的应用程序中实现创建和恢复搜索会话的功能之前，请问自己以下问题：

  1. 您的应用程序是否通常运行长时间的操作？ 例如，用户从冷存储中存储的数据生成仪表板或 Canvas 报告是有意义的。但是，在编辑单个可视化时，最好使用热数据或温数据的较短时间范围。
  2. 您的应用程序恢复搜索会话是否有意义？ 例如，您可能想要恢复在 Discover 中发现的旧文档的有趣过滤器配置。然而，单个 Lens 或地图可视化在特定仪表板上下文之外可能用处不大。
  3. 在您的应用程序上下文中，什么是搜索会话？ 虽然 Discover 和 Dashboard 每次时间范围或过滤器更改时，或者当用户点击刷新时都会启动新的搜索会话，但您可以不同地管理您的会话。例如，如果您的应用程序有标签页，您可以将来自多个标签页的搜索组合到一个搜索会话中。您必须能够清楚地定义用于创建搜索会话的状态。状态是指任何可能改变发送到 Elasticsearch 的查询的设置。

回答这些问题后，请继续在您的应用程序中实现以下代码片段。

在插件的 start 生命周期方法中，调用 enableStorage 方法。此方法有助于 Session Service 收集在用户请求时保存搜索会话所需的信息，并构建恢复状态

export class MyPlugin implements Plugin {
  public start(core: CoreStart, { data }: MyPluginStartDependencies) {
    const sessionRestorationDataProvider: SearchSessionInfoProvider = {
      data,
      getDashboard,
    };

    data.search.session.enableStorage({
      getName: async () => {
        // return the name you want to give the saved Search Session
        return `MyApp_${Math.random()}`;
      },
      getLocatorData: async () => {
        return {
          id: MY_LOCATOR,
          initialState: getLocatorParams({ ...deps, shouldRestoreSearchSession: false }),
          restoreState: getLocatorParams({ ...deps, shouldRestoreSearchSession: true }),
        };
      },
    });
  }
}
注意：搜索会话恢复状态

搜索会话的恢复状态可能与创建它时使用的初始状态不同。例如，初始状态可能包含相对日期，而在恢复状态中，这些日期必须转换为绝对日期。阅读有关 NowProvider 的更多信息。

重要：搜索会话指示器

调用 enableStorage 还将在解决方案的 chrome 组件中启用 Search Session Indicator 组件。Search Session Indicator 是一个小的按钮，默认用于吸引用户并保存新的搜索会话。要实现您自己的 UI，请联系 Kibana 应用程序服务团队以解耦此行为。

确保在您之前定义的状态发生变化时调用 start。

function onSearchSessionConfigChange() {
  this.searchSessionId = data.search.sessions.start();
}

将 searchSessionId 传递给应用程序中的每个 search 调用。如果您使用 Embeddables，请将 searchSessionId 作为 input 传递。

如果您无法直接传递 searchSessionId，则可以从服务中检索它。

const currentSearchSessionId = data.search.sessions.getSessionId();

创建一个新的搜索会话会清除前一个会话。当您的应用程序被销毁时，您必须显式地 clear 搜索会话

function onDestroy() {
  data.search.session.clear();
}

如果您不调用 clear，在开发时您会在控制台中看到警告。但是，在生产环境中运行时，您会得到一个致命错误。这样做是为了避免不相关的搜索请求泄漏到因疏忽而保持打开状态的现有搜索会话中。

集成的最后一步是恢复现有的搜索会话。searchSessionId 参数和其余恢复状态通过 URL 传递给应用程序。非 URL 支持计划在未来的版本中实现。

如果您检测到 URL 中存在 searchSessionId 参数，请调用 restore 方法，而不是调用 start。上面的示例现在将变为

function onSearchSessionConfigChange(searchSessionIdFromUrl?: string) {
  if (searchSessionIdFromUrl) {
    data.search.sessions.restore(searchSessionIdFromUrl);
  } else {
    data.search.sessions.start();
  }
}

一旦您 restore 了会话，只要所有 search 请求都使用相同的 searchSessionId 运行，搜索会话就应该被无缝恢复。

待定

© . This website operates independently and is not affiliated with or endorsed by Elasticsearch B.V. All brand names, logos, and trademarks are the property of their respective owners.