|
| 1 | +package org.apache.hadoop.ozone.om; |
| 2 | + |
| 3 | +import com.google.protobuf.Message; |
| 4 | +import org.apache.hadoop.ipc_.AlignmentContext; |
| 5 | +import org.apache.hadoop.ipc_.RetriableException; |
| 6 | +import org.apache.hadoop.ipc_.protobuf.RpcHeaderProtos.RpcRequestHeaderProto; |
| 7 | +import org.apache.hadoop.ipc_.protobuf.RpcHeaderProtos.RpcResponseHeaderProto; |
| 8 | +import org.apache.hadoop.ozone.OmUtils; |
| 9 | +import org.apache.hadoop.ozone.om.protocol.OzoneManagerProtocol; |
| 10 | +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; |
| 11 | +import org.apache.ratis.proto.RaftProtos.RaftPeerRole; |
| 12 | +import org.slf4j.Logger; |
| 13 | +import org.slf4j.LoggerFactory; |
| 14 | +import java.io.IOException; |
| 15 | +import java.util.concurrent.TimeUnit; |
| 16 | + |
| 17 | +/** |
| 18 | + * This is the server side implementation responsible for passing |
| 19 | + * state alignment info to clients. |
| 20 | + * </p> |
| 21 | + * Unlike HDFS's ClientNamenodeProtocol that has a RPC method for each |
| 22 | + * distinct call, OM (OzoneManagerService) only contains a single RPC method |
| 23 | + * (i.e. submitRequest(OMRequest)). Therefore, we need to query the OMRequest |
| 24 | + * to get the OMRequest parameter from the RPC method get the corresponding cmdType |
| 25 | + * to check whether we can the request to be run on non-leader OMs. |
| 26 | + */ |
| 27 | +public class OmAlignmentContext implements AlignmentContext { |
| 28 | + |
| 29 | + private static final Logger LOG = |
| 30 | + LoggerFactory.getLogger(OmAlignmentContext.class); |
| 31 | + /** |
| 32 | + * Estimated number of journal transactions a typical OM can execute |
| 33 | + * per second. The number is used to estimate how long a client's |
| 34 | + * RPC request will wait in the call queue before the Observer catches up |
| 35 | + * with its state id. |
| 36 | + */ |
| 37 | + private static final long ESTIMATED_TRANSACTIONS_PER_SECOND = 10000L; |
| 38 | + |
| 39 | + /** |
| 40 | + * The client wait time on an RPC request is composed of |
| 41 | + * the server execution time plus the communication time. |
| 42 | + * This is an expected fraction of the total wait time spent on |
| 43 | + * server execution. |
| 44 | + */ |
| 45 | + private static final float ESTIMATED_SERVER_TIME_MULTIPLIER = 0.8f; |
| 46 | + |
| 47 | + private final OzoneManager ozoneManager; |
| 48 | + |
| 49 | + /** |
| 50 | + * Server side constructor. |
| 51 | + * @param ozoneManager server side state provider |
| 52 | + */ |
| 53 | + OmAlignmentContext(OzoneManager ozoneManager) { |
| 54 | + this.ozoneManager = ozoneManager; |
| 55 | + } |
| 56 | + |
| 57 | + /** |
| 58 | + * Server side implementation for providing state alignment info in responses. |
| 59 | + */ |
| 60 | + @Override |
| 61 | + public void updateResponseState(RpcResponseHeaderProto.Builder header) { |
| 62 | + // Using getCorrectLastAppliedOrWrittenTxId will acquire the lock on |
| 63 | + // FSEditLog. This is needed so that ANN will return the correct state id |
| 64 | + // it currently has. But this may not be necessary for Observer, may want |
| 65 | + // revisit for optimization. Same goes to receiveRequestState. |
| 66 | + header.setStateId(getLastSeenStateId()); |
| 67 | + } |
| 68 | + |
| 69 | + /** |
| 70 | + * Server side implementation only provides state alignment info. |
| 71 | + * It does not receive state alignment info therefore this does nothing. |
| 72 | + */ |
| 73 | + @Override |
| 74 | + public void receiveResponseState(RpcResponseHeaderProto header) { |
| 75 | + // Do nothing. |
| 76 | + } |
| 77 | + |
| 78 | + /** |
| 79 | + * Server side implementation only receives state alignment info. |
| 80 | + * It does not build RPC requests therefore this does nothing. |
| 81 | + */ |
| 82 | + @Override |
| 83 | + public void updateRequestState(RpcRequestHeaderProto.Builder header) { |
| 84 | + // Do nothing. |
| 85 | + } |
| 86 | + |
| 87 | + /** |
| 88 | + * Server-side implementation for processing state alignment info in |
| 89 | + * requests. |
| 90 | + * For Follower/Listener it compares the client and the server states and determines |
| 91 | + * if it makes sense to wait until the server catches up with the client |
| 92 | + * state. If not the server throws RetriableException so that the client |
| 93 | + * could retry the call according to the retry policy with another Follower/Listener |
| 94 | + * or the Leader. |
| 95 | + * |
| 96 | + * @param header The RPC request header. |
| 97 | + * @param clientWaitTime time in milliseconds indicating how long client |
| 98 | + * waits for the server response. It is used to verify if the client's |
| 99 | + * state is too far ahead of the server's |
| 100 | + * @return the minimum of the state ids of the client or the server. |
| 101 | + * @throws RetriableException if Observer is too far behind. |
| 102 | + */ |
| 103 | + @Override |
| 104 | + public long receiveRequestState(RpcRequestHeaderProto header, |
| 105 | + long clientWaitTime) throws IOException { |
| 106 | + RaftPeerRole selfRole; |
| 107 | + if (ozoneManager.getOmRatisServer() == null) { |
| 108 | + selfRole = RaftPeerRole.LEADER; |
| 109 | + } else { |
| 110 | + selfRole = ozoneManager.getSelfRole(ozoneManager.getOmRatisServer().getLeaderId()); |
| 111 | + } |
| 112 | + |
| 113 | + if (!header.hasStateId() && !RaftPeerRole.LEADER.equals(selfRole)) { |
| 114 | + // This could happen if client configured with non-follower proxy provider |
| 115 | + // (e.g., ConfiguredFailoverProxyProvider) is accessing a cluster with. |
| 116 | + // In this case, we should let the client failover to the |
| 117 | + // leader node, rather than potentially serving stale result (client |
| 118 | + // stateId is 0 if not set). |
| 119 | + throw new IOException("Node received request without " |
| 120 | + + "stateId. This mostly likely is because client is not configured " |
| 121 | + + "with ObserverReadProxyProvider"); |
| 122 | + } |
| 123 | + long serverStateId = getLastSeenStateId(); |
| 124 | + long clientStateId = header.getStateId(); |
| 125 | + LOG.trace("Client State ID= {} and Server State ID= {}", |
| 126 | + clientStateId, serverStateId); |
| 127 | + |
| 128 | + if (clientStateId > serverStateId && |
| 129 | + RaftPeerRole.LEADER.equals(selfRole)) { |
| 130 | + LOG.warn("The client stateId: {} is greater than " |
| 131 | + + "the server stateId: {} This is unexpected. " |
| 132 | + + "Resetting client stateId to server stateId", |
| 133 | + clientStateId, serverStateId); |
| 134 | + return serverStateId; |
| 135 | + } |
| 136 | + if ((RaftPeerRole.FOLLOWER.equals(selfRole) || RaftPeerRole.LISTENER.equals(selfRole)) && |
| 137 | + clientStateId - serverStateId > |
| 138 | + ESTIMATED_TRANSACTIONS_PER_SECOND |
| 139 | + * TimeUnit.MILLISECONDS.toSeconds(clientWaitTime) |
| 140 | + * ESTIMATED_SERVER_TIME_MULTIPLIER) { |
| 141 | + throw new RetriableException( |
| 142 | + "Follower / Listener Node is too far behind: serverStateId = " |
| 143 | + + serverStateId + " clientStateId = " + clientStateId); |
| 144 | + } |
| 145 | + return clientStateId; |
| 146 | + } |
| 147 | + |
| 148 | + @Override |
| 149 | + public long getLastSeenStateId() { |
| 150 | + return ozoneManager.getOmRatisServer().getLastAppliedTermIndex().getIndex(); |
| 151 | + } |
| 152 | + |
| 153 | + @Override |
| 154 | + public boolean isCoordinatedCall(String protocolName, String methodName, Message payload) { |
| 155 | + // See OzoneManagerProtocolPB @ProtocolInfo annotation for the protocol name |
| 156 | + if (!protocolName.equals(OzoneManagerProtocol.class.getCanonicalName()) || |
| 157 | + !(methodName.equals("submitRequest")) || |
| 158 | + !(payload instanceof OMRequest)) { |
| 159 | + return false; |
| 160 | + } |
| 161 | + |
| 162 | + OMRequest omRequest = (OMRequest) payload; |
| 163 | + |
| 164 | + return OmUtils.isReadOnly(omRequest); |
| 165 | + } |
| 166 | + |
| 167 | +} |
0 commit comments