Merge pull request #3777 from akto-api-security/feature/logs_async

Feature/logs async

Author: ayushaga14

.cursor/rules/akto-core-rules.mdc

@@ -0,0 +1,7 @@
+---
+alwaysApply: true
+---
+
+- DO NOT REPEAT CODE, EVER
+- IF ANY CODE/FUNCTION/BLOCK CAN BE REUSED, REUSE IT.
+- IF ANY PIECE OF CODE HAS BEEN CREATED BY YOU, BUT NOT BEING USED, DELETE IT.

apps/mini-runtime/src/main/java/com/akto/hybrid_runtime/Main.java

@@ -577,7 +577,7 @@ private static void bulkParseTrafficToResponseParams(long lastSyncOffset, Consum
                 printL(r.value());
                 if(LOG_DEBUG_RECORDS > 0){
                     LOG_DEBUG_RECORDS--;
-                    loggerMaker.infoAndAddToDb("Kafka record recieved" + r.value());
+                    loggerMaker.infoAndAddToDb("Kafka record received" + r.value());
                 }
                 AllMetrics.instance.setRuntimeKafkaRecordCount(1);
                 AllMetrics.instance.setRuntimeKafkaRecordSize(r.value().length());
@@ -601,7 +601,7 @@ private static void bulkParseTrafficToResponseParams(long lastSyncOffset, Consum
 
                 httpResponseParams = HttpCallParser.parseKafkaMessage(r.value());
                 if (httpResponseParams == null) {
-                    loggerMaker.error("httpresponse params was skipped due to invalid json requestBody");
+                    loggerMaker.error("HttpResponse params was skipped due to invalid json requestBody");
                     continue;
                 }
 
@@ -759,8 +759,6 @@ private static void sendToCentralKafka(String centralKafkaTopicName, List<HttpRe
         }
     }
 
-
-
     private static AccountInfo refreshAccountInfo(Map<Integer, AccountInfo> accountInfoMap, int accountIdInt) {
         AccountInfo accountInfo = accountInfoMap.computeIfAbsent(accountIdInt, k -> new AccountInfo());
 
@@ -817,8 +815,8 @@ public static List<HttpResponseParams> filterBasedOnHeaders(List<HttpResponsePar
                 accWiseResponse = accWiseResponseFiltered;
             }
 
-            Map<String, Map<Pattern, String>> apiCollectioNameMapper = accountSettings.convertApiCollectionNameMapperToRegex();
-            changeTargetCollection(apiCollectioNameMapper, accWiseResponse);
+            Map<String, Map<Pattern, String>> apiCollectionNameMapper = accountSettings.convertApiCollectionNameMapperToRegex();
+            changeTargetCollection(apiCollectionNameMapper, accWiseResponse);
         }
 
         return accWiseResponse;

libs/utils/src/main/java/com/akto/data_actor/ClientActor.java

@@ -98,6 +98,20 @@ public class ClientActor extends DataActor {
     private static final CodecRegistry codecRegistry = DaoInit.createCodecRegistry();
     public static final String CYBORG_URL = "https://cyborg.akto.io";
     private static ExecutorService threadPool = Executors.newFixedThreadPool(maxConcurrentBatchWrites);
+    private static final int maxConcurrentLogWrites = getMaxConcurrentLogWrites();
+    final private static ExecutorService logThreadPool = Executors.newFixedThreadPool(maxConcurrentLogWrites);
+
+    private static int getMaxConcurrentLogWrites() {
+        String envValue = System.getenv("MAX_CONCURRENT_LOG_WRITES");
+        if (envValue != null) {
+            try {
+                return Integer.parseInt(envValue);
+            } catch (NumberFormatException e) {
+                // fall through to default
+            }
+        }
+        return 50;
+    }
 
     /**
      * Dedicated thread pool for agent traffic log HTTP writes.
@@ -1257,44 +1271,37 @@ public void createCollectionSimpleForVpc(int vxlanId, String vpcId, List<Collect
         }
     }
 
-    public void insertRuntimeLog(Log log) {
-        Map<String, List<String>> headers = buildHeaders();
-        BasicDBObject obj = new BasicDBObject();
-        BasicDBObject logObj = new BasicDBObject();
-        logObj.put("key", log.getKey());
-        logObj.put("log", log.getLog());
-        logObj.put("timestamp", log.getTimestamp());
-        obj.put("log", logObj);
-        OriginalHttpRequest request = new OriginalHttpRequest(url + "/insertRuntimeLog", "", "POST", obj.toString(), headers, "");
-        try {
-            OriginalHttpResponse response = ApiExecutor.sendRequestBackOff(request, true, null, false, null);
-            String responsePayload = response.getBody();
-            if (response.getStatusCode() != 200 || responsePayload == null) {
-                loggerMaker.info("non 2xx response in insertRuntimeLog");
-                return;
+    private void insertLogAsync(Log log, String endpoint) {
+        logThreadPool.submit(() -> {
+            try {
+                Map<String, List<String>> headers = buildHeaders();
+                BasicDBObject obj = new BasicDBObject();
+                    BasicDBObject logObj = new BasicDBObject();
+                    logObj.put("key", log.getKey());
+                    logObj.put("log", log.getLog());
+                    logObj.put("timestamp", log.getTimestamp());
+                    obj.put("log", logObj);
+                OriginalHttpRequest request = new OriginalHttpRequest(url + endpoint, "", "POST", obj.toString(), headers, "");
+                OriginalHttpResponse response = ApiExecutor.sendRequestBackOff(request, true, null, false, null);
+                if (response.getStatusCode() != 200 || response.getBody() == null) {
+                    loggerMaker.info("non 2xx response in " + endpoint);
+                }
+            } catch (Exception e) {
+                loggerMaker.error("error in " + endpoint + ": " + e);
             }
-        } catch (Exception e) {
-            loggerMaker.error("error in insertRuntimeLog" + e);
-            return;
-        }
+        });
+    }
+
+    public void insertRuntimeLog(Log log) {
+        insertLogAsync(log, "/insertRuntimeLog");
     }
 
     public void insertAnalyserLog(Log log) {
-        Map<String, List<String>> headers = buildHeaders();
-        BasicDBObject obj = new BasicDBObject();
-        obj.put("log", log);
-        OriginalHttpRequest request = new OriginalHttpRequest(url + "/insertAnalyserLog", "", "POST", obj.toString(), headers, "");
-        try {
-            OriginalHttpResponse response = ApiExecutor.sendRequestBackOff(request, true, null, false, null);
-            String responsePayload = response.getBody();
-            if (response.getStatusCode() != 200 || responsePayload == null) {
-                loggerMaker.errorAndAddToDb("non 2xx response in insertAnalyserLog", LoggerMaker.LogDb.RUNTIME);
-                return;
-            }
-        } catch (Exception e) {
-            loggerMaker.errorAndAddToDb("error in insertAnalyserLog" + e, LoggerMaker.LogDb.RUNTIME);
-            return;
-        }
+        insertLogAsync(log, "/insertAnalyserLog");
+    }
+
+    public void insertTestingLog(Log log) {
+        insertLogAsync(log, "/insertTestingLog");
     }
 
     public void modifyHybridSaasSetting(boolean isHybridSaas) {
@@ -3313,28 +3320,6 @@ public void modifyHybridTestingSetting(boolean hybridTestingEnabled) {
         }
     }
 
-    public void insertTestingLog(Log log) {
-        Map<String, List<String>> headers = buildHeaders();
-        BasicDBObject obj = new BasicDBObject();
-        BasicDBObject logObj = new BasicDBObject();
-        logObj.put("key", log.getKey());
-        logObj.put("log", log.getLog());
-        logObj.put("timestamp", log.getTimestamp());
-        obj.put("log", logObj);
-        OriginalHttpRequest request = new OriginalHttpRequest(url + "/insertTestingLog", "", "POST", obj.toString(), headers, "");
-        try {
-            OriginalHttpResponse response = ApiExecutor.sendRequestBackOff(request, true, null, false, null);
-            String responsePayload = response.getBody();
-            if (response.getStatusCode() != 200 || responsePayload == null) {
-                loggerMaker.info("non 2xx response in insertTestingLog");
-                return;
-            }
-        } catch (Exception e) {
-            loggerMaker.error("error in insertTestingLog" + e);
-            return;
-        }
-    }
-
     public void bulkWriteDependencyNodes(List<DependencyNode> dependencyNodeList) {
         BasicDBObject obj = new BasicDBObject();
         obj.put("dependencyNodeList", dependencyNodeList);