/**
 * Constructs the Apollo data source
 *
 * @param namespaceName        the namespace name in Apollo, should not be null or empty
 * @param ruleKey              the rule key in the namespace, should not be null or empty
 * @param defaultRuleValue     the default rule value when the ruleKey is not found or any error
 *                             occurred
 * @param parser               the parser to transform string configuration to actual flow rules
 */
public ApolloDataSource(String namespaceName, String ruleKey, String defaultRuleValue,
                        Converter<String, T> parser) {
    super(parser);

    Preconditions.checkArgument(!Strings.isNullOrEmpty(namespaceName), "Namespace name could not be null or empty");
    Preconditions.checkArgument(!Strings.isNullOrEmpty(ruleKey), "RuleKey could not be null or empty!");

    this.ruleKey = ruleKey;
    this.defaultRuleValue = defaultRuleValue;

    this.config = ConfigService.getConfig(namespaceName);

    initialize();

    RecordLog.info(String.format("Initialized rule for namespace: %s, rule key: %s", namespaceName, ruleKey));
}
 

private static boolean passClusterCheck(FlowRule rule, Context context, DefaultNode node, int acquireCount,
                                        boolean prioritized) {
    try {
        TokenService clusterService = pickClusterService();
        if (clusterService == null) {
            return fallbackToLocalOrPass(rule, context, node, acquireCount, prioritized);
        }
        long flowId = rule.getClusterConfig().getFlowId();
        TokenResult result = clusterService.requestToken(flowId, acquireCount, prioritized);
        return applyTokenResult(result, rule, context, node, acquireCount, prioritized);
        // If client is absent, then fallback to local mode.
    } catch (Throwable ex) {
        RecordLog.warn("[FlowRuleChecker] Request cluster token unexpected failed", ex);
    }
    // Fallback to local flow control when token client or server for this rule is not available.
    // If fallback is not enabled, then directly pass.
    return fallbackToLocalOrPass(rule, context, node, acquireCount, prioritized);
}
 

private void connect(Bootstrap b) {
    if (currentState.compareAndSet(ClientConstants.CLIENT_STATUS_OFF, ClientConstants.CLIENT_STATUS_PENDING)) {
        b.connect(host, port)
            .addListener(new GenericFutureListener<ChannelFuture>() {
            @Override
            public void operationComplete(ChannelFuture future) {
                if (future.cause() != null) {
                    RecordLog.warn(
                        String.format("[NettyTransportClient] Could not connect to <%s:%d> after %d times",
                            host, port, failConnectedTime.get()), future.cause());
                    failConnectedTime.incrementAndGet();
                    channel = null;
                } else {
                    failConnectedTime.set(0);
                    channel = future.channel();
                    RecordLog.info(
                        "[NettyTransportClient] Successfully connect to server <" + host + ":" + port + ">");
                }
            }
        });
    }
}
 

@Override
public void stop() {
    // If still initializing, wait for ready.
    while (currentState.get() == SERVER_STATUS_STARTING) {
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            // Ignore.
        }
    }

    if (currentState.compareAndSet(SERVER_STATUS_STARTED, SERVER_STATUS_OFF)) {
        try {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
            connectionPool.shutdownAll();

            failedTimes.set(0);

            RecordLog.info("[NettyTransportServer] Sentinel token server stopped");
        } catch (Exception ex) {
            RecordLog.warn("[NettyTransportServer] Failed to stop token server (port=" + port + ")", ex);
        }
    }
}
 

@Override
public ClusterResponse decode(ByteBuf source) {
    if (source.readableBytes() >= 6) {
        int xid = source.readInt();
        int type = source.readByte();
        int status = source.readByte();

        EntityDecoder<ByteBuf, ?> decoder = ResponseDataDecodeRegistry.getDecoder(type);
        if (decoder == null) {
            RecordLog.warn("Unknown type of response data decoder: {0}", type);
            return null;
        }

        Object data;
        if (source.readableBytes() == 0) {
            data = null;
        } else {
            data = decoder.decode(source);
        }

        return new ClusterResponse<>(xid, type, status, data);
    }
    return null;
}
 

private static void resolveInstance() {
    RequestEntityWriter writer = SpiLoader.loadFirstInstance(RequestEntityWriter.class);
    if (writer == null) {
        RecordLog.warn("[ClientEntityCodecProvider] No existing request entity writer, resolve failed");
    } else {
        requestEntityWriter = writer;
        RecordLog.info("[ClientEntityCodecProvider] Request entity writer resolved: " + requestEntityWriter.getClass().getCanonicalName());
    }
    ResponseEntityDecoder decoder = SpiLoader.loadFirstInstance(ResponseEntityDecoder.class);
    if (decoder == null) {
        RecordLog.warn("[ClientEntityCodecProvider] No existing response entity decoder, resolve failed");
    } else {
        responseEntityDecoder = decoder;
        RecordLog.info("[ClientEntityCodecProvider] Response entity decoder resolved: " + responseEntityDecoder.getClass().getCanonicalName());
    }
}
 

@Override
public void configUpdate(List<SystemRule> rules) {
    restoreSetting();
    // systemRules = rules;
    if (rules != null && rules.size() >= 1) {
        for (SystemRule rule : rules) {
            loadSystemConf(rule);
        }
    } else {
        checkSystemStatus.set(false);
    }

    RecordLog.info(String.format("[SystemRuleManager] Current system check status: %s, "
            + "highestSystemLoad: %e, "
            + "highestCpuUsage: %e, "
            + "maxRt: %d, "
            + "maxThread: %d, "
            + "maxQps: %e",
        checkSystemStatus.get(),
        highestSystemLoad,
        highestCpuUsage,
        maxRt,
        maxThread,
        qps));
}
 

private void initNewConnection() {
    if (transportClient != null) {
        return;
    }
    String host = ClusterClientConfigManager.getServerHost();
    int port = ClusterClientConfigManager.getServerPort();
    if (StringUtil.isBlank(host) || port <= 0) {
        return;
    }

    try {
        this.transportClient = new NettyTransportClient(host, port);
        this.serverDescriptor = new TokenServerDescriptor(host, port);
        RecordLog.info("[DefaultClusterTokenClient] New client created: " + serverDescriptor);
    } catch (Exception ex) {
        RecordLog.warn("[DefaultClusterTokenClient] Failed to initialize new token client", ex);
    }
}
 

private void changeServer(/*@Valid*/ ClusterClientAssignConfig config) {
    if (serverEqual(serverDescriptor, config)) {
        return;
    }
    try {
        if (transportClient != null) {
            transportClient.stop();
        }
        // Replace with new, even if the new client is not ready.
        this.transportClient = new NettyTransportClient(config.getServerHost(), config.getServerPort());
        this.serverDescriptor = new TokenServerDescriptor(config.getServerHost(), config.getServerPort());
        startClientIfScheduled();
        RecordLog.info("[DefaultClusterTokenClient] New client created: " + serverDescriptor);
    } catch (Exception ex) {
        RecordLog.warn("[DefaultClusterTokenClient] Failed to change remote token server", ex);
    }
}
 

private static boolean passClusterCheck(FlowRule rule, Context context, DefaultNode node, int acquireCount,
                                        boolean prioritized) {
    try {
        TokenService clusterService = pickClusterService();
        if (clusterService == null) {
            return fallbackToLocalOrPass(rule, context, node, acquireCount, prioritized);
        }
        long flowId = rule.getClusterConfig().getFlowId();
        TokenResult result = clusterService.requestToken(flowId, acquireCount, prioritized);
        return applyTokenResult(result, rule, context, node, acquireCount, prioritized);
        // If client is absent, then fallback to local mode.
    } catch (Throwable ex) {
        RecordLog.warn("[FlowRuleChecker] Request cluster token unexpected failed", ex);
    }
    // Fallback to local flow control when token client or server for this rule is not available.
    // If fallback is not enabled, then directly pass.
    return fallbackToLocalOrPass(rule, context, node, acquireCount, prioritized);
}
 

private static int resolvePort() {
    final int defaultPort = SentinelEnvoyRlsConstants.DEFAULT_GRPC_PORT;
    // Order: system env > property
    String portStr = Optional.ofNullable(System.getenv(SentinelEnvoyRlsConstants.GRPC_PORT_ENV_KEY))
        .orElse(SentinelConfig.getConfig(SentinelEnvoyRlsConstants.GRPC_PORT_PROPERTY_KEY));
    if (StringUtil.isBlank(portStr)) {
        return defaultPort;
    }
    try {
        int port = Integer.parseInt(portStr);
        if (port <= 0 || port > 65535) {
            RecordLog.warn("[SentinelEnvoyRlsServer] Invalid port <" + portStr + ">, using default" + defaultPort);
            return defaultPort;
        }
        return port;
    } catch (Exception ex) {
        RecordLog.warn("[SentinelEnvoyRlsServer] Failed to resolve port, using default " + defaultPort);
        System.err.println("[SentinelEnvoyRlsServer] Failed to resolve port, using default " + defaultPort);
        return defaultPort;
    }
}
 

@Override
public synchronized void configUpdate(List<SystemRule> rules) {
    restoreSetting();
    // systemRules = rules;
    if (rules != null && rules.size() >= 1) {
        for (SystemRule rule : rules) {
            loadSystemConf(rule);
        }
    } else {
        checkSystemStatus.set(false);
    }

    RecordLog.info(String.format("[SystemRuleManager] Current system check status: %s, "
            + "highestSystemLoad: %e, "
            + "highestCpuUsage: %e, "
            + "maxRt: %d, "
            + "maxThread: %d, "
            + "maxQps: %e",
        checkSystemStatus.get(),
        highestSystemLoad,
        highestCpuUsage,
        maxRt,
        maxThread,
        qps));
}
 

private Method findMethod(boolean mustStatic, Class<?> clazz, String name, Class<?> returnType,
                          Class<?>... parameterTypes) {
    Method[] methods = clazz.getDeclaredMethods();
    for (Method method : methods) {
        if (name.equals(method.getName()) && checkStatic(mustStatic, method)
            && returnType.isAssignableFrom(method.getReturnType())
            && Arrays.equals(parameterTypes, method.getParameterTypes())) {

            RecordLog.info("Resolved method [{}] in class [{}]", name, clazz.getCanonicalName());
            return method;
        }
    }
    // Current class not found, find in the super classes recursively.
    Class<?> superClass = clazz.getSuperclass();
    if (superClass != null && !Object.class.equals(superClass)) {
        return findMethod(mustStatic, superClass, name, returnType, parameterTypes);
    } else {
        String methodType = mustStatic ? " static" : "";
        RecordLog.warn("Cannot find{} method [{}] in class [{}] with parameters {}",
            methodType, name, clazz.getCanonicalName(), Arrays.toString(parameterTypes));
        return null;
    }
}
 

private static void resolveInstance() {
    ResponseEntityWriter writer = SpiLoader.loadFirstInstance(ResponseEntityWriter.class);
    if (writer == null) {
        RecordLog.warn("[ServerEntityCodecProvider] No existing response entity writer, resolve failed");
    } else {
        responseEntityWriter = writer;
        RecordLog.info(
            "[ServerEntityCodecProvider] Response entity writer resolved: " + responseEntityWriter.getClass()
                .getCanonicalName());
    }
    RequestEntityDecoder decoder = SpiLoader.loadFirstInstance(RequestEntityDecoder.class);
    if (decoder == null) {
        RecordLog.warn("[ServerEntityCodecProvider] No existing request entity decoder, resolve failed");
    } else {
        requestEntityDecoder = decoder;
        RecordLog.info(
            "[ServerEntityCodecProvider] Request entity decoder resolved: " + requestEntityDecoder.getClass()
                .getCanonicalName());
    }
}
 

@Override
public ClusterRequest decode(ByteBuf source) {
    if (source.readableBytes() >= 5) {
        int xid = source.readInt();
        int type = source.readByte();

        EntityDecoder<ByteBuf, ?> dataDecoder = RequestDataDecodeRegistry.getDecoder(type);
        if (dataDecoder == null) {
            RecordLog.warn("Unknown type of request data decoder: {0}", type);
            return null;
        }

        Object data;
        if (source.readableBytes() == 0) {
            data = null;
        } else {
            data = dataDecoder.decode(source);
        }

        return new ClusterRequest<>(xid, type, data);
    }
    return null;
}
 

@Override
public CommandResponse<String> handle(CommandRequest request) {
    String data = request.getParam("data");
    if (StringUtil.isBlank(data)) {
        return CommandResponse.ofFailure(new IllegalArgumentException("Bad data"));
    }
    try {
        data = URLDecoder.decode(data, "utf-8");
    } catch (Exception e) {
        RecordLog.info("Decode gateway API definition data error", e);
        return CommandResponse.ofFailure(e, "decode gateway API definition data error");
    }

    RecordLog.info("[API Server] Receiving data change (type: gateway API definition): {0}", data);

    String result = SUCCESS_MSG;

    Set<ApiDefinition> apiDefinitions = parseJson(data);
    GatewayApiDefinitionManager.loadApiDefinitions(apiDefinitions);

    return CommandResponse.ofSuccess(result);
}
 

private Method findMethod(boolean mustStatic, Class<?> clazz, String name, Class<?> returnType,
                          Class<?>... parameterTypes) {
    Method[] methods = clazz.getDeclaredMethods();
    for (Method method : methods) {
        if (name.equals(method.getName()) && checkStatic(mustStatic, method)
                && returnType.isAssignableFrom(method.getReturnType())
                && Arrays.equals(parameterTypes, method.getParameterTypes())) {

            RecordLog.info("Resolved method [{0}] in class [{1}]", name, clazz.getCanonicalName());
            return method;
        }
    }
    // Current class not found, find in the super classes recursively.
    Class<?> superClass = clazz.getSuperclass();
    if (superClass != null && !Object.class.equals(superClass)) {
        return findMethod(mustStatic, superClass, name, returnType, parameterTypes);
    } else {
        String methodType = mustStatic ? " static" : "";
        RecordLog.warn("Cannot find{0} method [{1}] in class [{2}] with parameters {3}",
                methodType, name, clazz.getCanonicalName(), Arrays.toString(parameterTypes));
        return null;
    }
}
 

private void connect(Bootstrap b) {
    if (currentState.compareAndSet(ClientConstants.CLIENT_STATUS_OFF, ClientConstants.CLIENT_STATUS_PENDING)) {
        b.connect(host, port)
            .addListener(new GenericFutureListener<ChannelFuture>() {
            @Override
            public void operationComplete(ChannelFuture future) {
                if (future.cause() != null) {
                    RecordLog.warn(
                        String.format("[NettyTransportClient] Could not connect to <%s:%d> after %d times",
                            host, port, failConnectedTime.get()), future.cause());
                    failConnectedTime.incrementAndGet();
                    channel = null;
                } else {
                    failConnectedTime.set(0);
                    channel = future.channel();
                    RecordLog.info(
                        "[NettyTransportClient] Successfully connect to server <" + host + ":" + port + ">");
                }
            }
        });
    }
}
 

/**
 * Listen to the {@link SentinelProperty} for Envoy RLS rules. The property is the source of {@link EnvoyRlsRule}.
 *
 * @param property the property to listen
 */
public static void register2Property(SentinelProperty<List<EnvoyRlsRule>> property) {
    AssertUtil.notNull(property, "property cannot be null");
    synchronized (PROPERTY_LISTENER) {
        RecordLog.info("[EnvoyRlsRuleManager] Registering new property to Envoy rate limit service rule manager");
        currentProperty.removeListener(PROPERTY_LISTENER);
        property.addListener(PROPERTY_LISTENER);
        currentProperty = property;
    }
}
 

@Override
public void configLoad(List<ParamFlowRule> list) {
    Map<String, List<ParamFlowRule>> rules = aggregateAndPrepareParamRules(list);
    if (rules != null) {
        paramFlowRules.clear();
        paramFlowRules.putAll(rules);
    }
    RecordLog.info("[ParamFlowRuleManager] Parameter flow rules received: " + paramFlowRules);
}
 

public void listenChanged() {
    try {
        Object newValue = dataSource.loadConfig();
        dataSource.getProperty().updateValue(newValue);
    } catch (Exception e) {
        RecordLog.warn("[SpringConfigListener] load config error: ", e);
    }
}
 

public AbstractApiMatcher(ApiDefinition apiDefinition) {
    AssertUtil.notNull(apiDefinition, "apiDefinition cannot be null");
    AssertUtil.assertNotBlank(apiDefinition.getApiName(), "apiName cannot be empty");
    this.apiName = apiDefinition.getApiName();
    this.apiDefinition = apiDefinition;

    try {
        initializeMatchers();
    } catch (Exception ex) {
        RecordLog.warn("[GatewayApiMatcher] Failed to initialize internal matchers", ex);
    }
}
 

private static void resolveTokenClientInstance() {
    ClusterTokenClient resolvedClient = SpiLoader.loadFirstInstance(ClusterTokenClient.class);
    if (resolvedClient == null) {
        RecordLog.info(
            "[TokenClientProvider] No existing cluster token client, cluster client mode will not be activated");
    } else {
        client = resolvedClient;
        RecordLog.info(
            "[TokenClientProvider] Cluster token client resolved: " + client.getClass().getCanonicalName());
    }
}
 

/**
 * Get heartbeat interval in milliseconds.
 *
 * @return heartbeat interval in milliseconds if exists, or null if not configured or invalid config
 */
public static Long getHeartbeatIntervalMs() {
    String interval = SentinelConfig.getConfig(HEARTBEAT_INTERVAL_MS);
    try {
        return interval == null ? null : Long.parseLong(interval);
    } catch (Exception ex) {
        RecordLog.warn("[TransportConfig] Failed to parse heartbeat interval: " + interval);
        return null;
    }
}
 

private static void resolveInstance() {
    CommandCenter resolveCommandCenter = SpiLoader.loadHighestPriorityInstance(CommandCenter.class);

    if (resolveCommandCenter == null) {
        RecordLog.warn("[CommandCenterProvider] WARN: No existing CommandCenter found");
    } else {
        commandCenter = resolveCommandCenter;
        RecordLog.info("[CommandCenterProvider] CommandCenter resolved: " + resolveCommandCenter.getClass()
            .getCanonicalName());
    }
}
 

@Override
public void configLoad(List<AuthorityRule> value) {
    Map<String, Set<AuthorityRule>> rules = loadAuthorityConf(value);

    authorityRules.clear();
    if (rules != null) {
        authorityRules.putAll(rules);
    }
    RecordLog.info("[AuthorityRuleManager] Load authority rules: " + authorityRules);
}
 

/**
 * Update the {@link #SAMPLE_COUNT}. All {@link ClusterNode}s will be reset if newSampleCount
 * is different from {@link #SAMPLE_COUNT}.
 *
 * @param newSampleCount New sample count to set. This value must be divisor of 1000.
 */
public static void updateSampleCount(int newSampleCount) {
    if (newSampleCount != SAMPLE_COUNT) {
        SAMPLE_COUNT = newSampleCount;
        ClusterBuilderSlot.resetClusterNodes();
    }
    RecordLog.info("SAMPLE_COUNT updated to: " + SAMPLE_COUNT);
}
 

@Override
public boolean sendHeartbeat() throws Exception {
    if (TransportConfig.getRuntimePort() <= 0) {
        RecordLog.info("[SimpleHttpHeartbeatSender] Command server port not initialized, won't send heartbeat");
        return false;
    }
    Tuple2<String, Integer> addrInfo = getAvailableAddress();
    if (addrInfo == null) {
        return false;
    }

    InetSocketAddress addr = new InetSocketAddress(addrInfo.r1, addrInfo.r2);
    SimpleHttpRequest request = new SimpleHttpRequest(addr, TransportConfig.getHeartbeatApiPath());
    request.setParams(heartBeat.generateCurrentMessage());
    try {
        SimpleHttpResponse response = httpClient.post(request);
        if (response.getStatusCode() == OK_STATUS) {
            return true;
        } else if (clientErrorCode(response.getStatusCode()) || serverErrorCode(response.getStatusCode())) {
            RecordLog.warn("[SimpleHttpHeartbeatSender] Failed to send heartbeat to " + addr
                + ", http status code: " + response.getStatusCode());
        }
    } catch (Exception e) {
        RecordLog.warn("[SimpleHttpHeartbeatSender] Failed to send heartbeat to " + addr, e);
    }
    return false;
}
 

@Override
public void configLoad(List<DegradeRule> conf) {
    Map<String, Set<DegradeRule>> rules = loadDegradeConf(conf);
    if (rules != null) {
        degradeRules.clear();
        degradeRules.putAll(rules);
    }
    RecordLog.info("[DegradeRuleManager] Degrade rules loaded: " + degradeRules);
}
 

/**
 * Remove cluster flow rule property for a specific namespace.
 *
 * @param namespace valid namespace
 */
public static void removeProperty(String namespace) {
    AssertUtil.notEmpty(namespace, "namespace cannot be empty");
    synchronized (UPDATE_LOCK) {
        NamespaceFlowProperty<FlowRule> property = PROPERTY_MAP.get(namespace);
        if (property != null) {
            property.getProperty().removeListener(property.getListener());
            PROPERTY_MAP.remove(namespace);
        }
        RecordLog.info("[ClusterFlowRuleManager] Removing property from cluster flow rule manager"
            + " for namespace <{}>", namespace);
    }
}