/* * SPDX-License-Identifier: Apache-2.0 * * The OpenSearch Contributors require contributions made to * this file be licensed under the Apache-2.0 license or a * compatible open source license. */ /* * Licensed to Elasticsearch under one or more contributor * license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright * ownership. Elasticsearch licenses this file to you under * the Apache License, Version 2.0 (the "License"); you may * not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, * software distributed under the License is distributed on an * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * KIND, either express or implied. See the License for the * specific language governing permissions and limitations * under the License. */ /* * Modifications Copyright OpenSearch Contributors. See * GitHub history for details. */ package org.opensearch.action.get; import org.opensearch.action.ActionListener; import org.opensearch.action.RoutingMissingException; import org.opensearch.action.support.ActionFilters; import org.opensearch.action.support.single.shard.TransportSingleShardAction; import org.opensearch.cluster.ClusterState; import org.opensearch.cluster.metadata.IndexNameExpressionResolver; import org.opensearch.cluster.routing.ShardIterator; import org.opensearch.cluster.service.ClusterService; import org.opensearch.common.inject.Inject; import org.opensearch.core.common.io.stream.Writeable; import org.opensearch.index.IndexService; import org.opensearch.index.get.GetResult; import org.opensearch.index.shard.IndexShard; import org.opensearch.core.index.shard.ShardId; import org.opensearch.indices.IndicesService; import org.opensearch.threadpool.ThreadPool; import org.opensearch.transport.TransportService; import java.io.IOException; /** * Performs the get operation. * * @opensearch.internal */ public class TransportGetAction extends TransportSingleShardAction { private final IndicesService indicesService; @Inject public TransportGetAction( ClusterService clusterService, TransportService transportService, IndicesService indicesService, ThreadPool threadPool, ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver ) { super( GetAction.NAME, threadPool, clusterService, transportService, actionFilters, indexNameExpressionResolver, GetRequest::new, ThreadPool.Names.GET ); this.indicesService = indicesService; } @Override protected boolean resolveIndex(GetRequest request) { return true; } @Override protected ShardIterator shards(ClusterState state, InternalRequest request) { return clusterService.operationRouting() .getShards( clusterService.state(), request.concreteIndex(), request.request().id(), request.request().routing(), request.request().preference() ); } @Override protected void resolveRequest(ClusterState state, InternalRequest request) { // update the routing (request#index here is possibly an alias) request.request().routing(state.metadata().resolveIndexRouting(request.request().routing(), request.request().index())); // Fail fast on the node that received the request. if (request.request().routing() == null && state.getMetadata().routingRequired(request.concreteIndex())) { throw new RoutingMissingException(request.concreteIndex(), request.request().id()); } } @Override protected void asyncShardOperation(GetRequest request, ShardId shardId, ActionListener listener) throws IOException { IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex()); IndexShard indexShard = indexService.getShard(shardId.id()); if (request.realtime()) { // we are not tied to a refresh cycle here anyway super.asyncShardOperation(request, shardId, listener); } else { indexShard.awaitShardSearchActive(b -> { try { super.asyncShardOperation(request, shardId, listener); } catch (Exception ex) { listener.onFailure(ex); } }); } } @Override protected GetResponse shardOperation(GetRequest request, ShardId shardId) { IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex()); IndexShard indexShard = indexService.getShard(shardId.id()); if (request.refresh() && !request.realtime()) { indexShard.refresh("refresh_flag_get"); } GetResult result = indexShard.getService() .get( request.id(), request.storedFields(), request.realtime(), request.version(), request.versionType(), request.fetchSourceContext() ); return new GetResponse(result); } @Override protected Writeable.Reader getResponseReader() { return GetResponse::new; } @Override protected String getExecutor(GetRequest request, ShardId shardId) { final ClusterState clusterState = clusterService.state(); if (clusterState.metadata().getIndexSafe(shardId.getIndex()).isSystem()) { return ThreadPool.Names.SYSTEM_READ; } else if (indicesService.indexServiceSafe(shardId.getIndex()).getIndexSettings().isSearchThrottled()) { return ThreadPool.Names.SEARCH_THROTTLED; } else { return super.getExecutor(request, shardId); } } }