Apache/InLong InLong Manager 支持配置 Flink 任务并发度/Adjust sort resources according to data scale心得

audit已经实现了对于InLong系统的Agent、DataProxy、Sort模块的入流量、出流量进行实时审计对账。 对账的粒度有分钟、小时、天三种粒度。

1. audit的数据缓存在org.apache.inlong.audit.cache的各个类中，有DayCache HalfHourCache等等
2. 请求audit数据的api在org.apache.inlong.audit.config.OpenApiConstants中，通过org.apache.inlong.audit.service.ApiService来调用。调用的时候注意参数的合法性，有合法性检查，以免请求不到数据。居然没有用controller，挺神奇的，贴一段代码上来，便于以后研究

private void initHttpServer() {

 private void initHttpServer() {
        int bindPort = Configuration.getInstance().get(KEY_HTTP_SERVER_BIND_PORT, DEFAULT_HTTP_SERVER_BIND_PORT);
        try {
            HttpServer server = HttpServer.create(new InetSocketAddress(bindPort),
                    Configuration.getInstance().get(KEY_API_BACKLOG_SIZE, DEFAULT_API_BACKLOG_SIZE));
            server.setExecutor(Executors.newFixedThreadPool(
                    Configuration.getInstance().get(KEY_API_THREAD_POOL_SIZE, DEFAULT_API_THREAD_POOL_SIZE)));
            server.createContext(Configuration.getInstance().get(KEY_API_DAY_PATH, DEFAULT_API_DAY_PATH),
                    new AuditHandler(DAY));
            server.createContext(Configuration.getInstance().get(KEY_API_HOUR_PATH, DEFAULT_API_HOUR_PATH),
                    new AuditHandler(HOUR));
            server.createContext(Configuration.getInstance().get(KEY_API_MINUTES_PATH, DEFAULT_API_MINUTES_PATH),
........省略类似的方法.........
            server.start();
            LOGGER.info("Init http server success. Bind port is: {}", bindPort);
        } catch (Exception e) {
            LOGGER.error("Init http server has exception!", e);
        }
    }

    int bindPort = Configuration.getInstance().get(KEY_HTTP_SERVER_BIND_PORT, DEFAULT_HTTP_SERVER_BIND_PORT);
    try {
        HttpServer server = HttpServer.create(new InetSocketAddress(bindPort),
                Configuration.getInstance().get(KEY_API_BACKLOG_SIZE, DEFAULT_API_BACKLOG_SIZE));
        server.setExecutor(Executors.newFixedThreadPool(
                Configuration.getInstance().get(KEY_API_THREAD_POOL_SIZE, DEFAULT_API_THREAD_POOL_SIZE)));
        server.createContext(Configuration.getInstance().get(KEY_API_DAY_PATH, DEFAULT_API_DAY_PATH),
                new AuditHandler(DAY));
        server.createContext(Configuration.getInstance().get(KEY_API_HOUR_PATH, DEFAULT_API_HOUR_PATH),
                new AuditHandler(HOUR));
        server.createContext(Configuration.getInstance().get(KEY_API_MINUTES_PATH, DEFAULT_API_MINUTES_PATH),

........省略类似的方法.........
server.start();
LOGGER.info("Init http server success. Bind port is: {}", bindPort);
} catch (Exception e) {
LOGGER.error("Init http server has exception!", e);
}
}

3. 请求方法样例，查询小时数据curl localhost:10080/audit/hour?startTime=2024-07-01T00:00:00&endTime=2024-07-07T23:59:59&inlongGroupId=testGroup&inlongStreamId=testStream&auditId=testAudit