Feature/s2s graph

apps/api-runtime/src/main/java/com/akto/runtime/APICatalogSync.java

@@ -8,6 +8,9 @@
 import com.akto.dao.context.Context;
 import com.akto.dao.monitoring.FilterYamlTemplateDao;
 import com.akto.dao.runtime_filters.AdvancedTrafficFiltersDao;
+import com.akto.data_actor.DataActor;
+import com.akto.data_actor.DataActorFactory;
+import com.akto.data_actor.DbActor;
 import com.akto.dao.filter.MergedUrlsDao;
 import com.akto.dto.*;
 import com.akto.dto.billing.SyncLimit;
@@ -27,6 +30,8 @@
 import com.akto.log.LoggerMaker;
 import com.akto.log.LoggerMaker.LogDb;
 import com.akto.util.filter.DictionaryFilter;
+import com.akto.runtime.APICatalogSync.ApiMergerResult;
+import com.akto.runtime.APICatalogSync.DbUpdateReturn;
 import com.akto.runtime.merge.MergeOnHostOnly;
 import com.akto.runtime.policies.AktoPolicyNew;
 import com.akto.task.Cluster;
@@ -66,6 +71,8 @@ public class APICatalogSync {
     public Map<Integer, APICatalog> dbState;
     public Map<Integer, APICatalog> delta;
     public AktoPolicyNew aktoPolicyNew;
+    public SvcToSvcGraphManager svcToSvcGraphManager = null;
+    public DataActor dataActor = DataActorFactory.fetchInstance();
     public Map<SensitiveParamInfo, Boolean> sensitiveParamInfoBooleanMap;
     public static boolean mergeAsyncOutside = true;
     public BloomFilter<CharSequence> existingAPIsInDb = BloomFilter.create(Funnels.stringFunnel(Charsets.UTF_8), 1_000_000, 0.001 );
@@ -88,6 +95,7 @@ public APICatalogSync(String userIdentifier, int thresh, boolean fetchAllSTI, bo
         mergedUrls = new HashSet<>();
         if (buildFromDb) {
             buildFromDB(false, fetchAllSTI);
+            this.svcToSvcGraphManager = SvcToSvcGraphManager.createFromEdgesAndNodes(dataActor);
         }
     }
 
@@ -214,6 +222,9 @@ public void computeDelta(URLAggregator origAggregator, boolean triggerTemplateGe
             for (HttpResponseParams responseParams: value) {
                 try {
                     aktoPolicyNew.process(responseParams);
+                    if (svcToSvcGraphManager != null) {
+                        svcToSvcGraphManager.processRecord(responseParams.getSvcToSvcGraphParams());
+                    }
                 } catch (Exception e) {
                     e.printStackTrace();
                     throw new RuntimeException(e);
@@ -1890,7 +1901,14 @@ public void syncWithDB(boolean syncImmediately, boolean fetchAllSTI, SyncLimit s
         }
 
         loggerMaker.infoAndAddToDb("starting build from db inside syncWithDb", LogDb.RUNTIME);
+
         buildFromDB(true, fetchAllSTI);
+        if (svcToSvcGraphManager != null){
+            svcToSvcGraphManager.updateWithNewDataAndReturnDelta(dataActor);            
+            if (svcToSvcGraphManager.getLastFetchFromDb() < Context.now() - 6 * 60 * 60) {
+                svcToSvcGraphManager = SvcToSvcGraphManager.createFromEdgesAndNodes(dataActor);
+            }
+        }
         loggerMaker.infoAndAddToDb("Finished syncing with db", LogDb.RUNTIME);
     }
 

apps/database-abstractor/src/main/java/com/akto/action/DbAction.java

@@ -11,6 +11,9 @@
 import com.akto.dto.bulk_updates.BulkUpdates;
 import com.akto.dto.bulk_updates.UpdatePayload;
 import com.akto.dto.filter.MergedUrls;
+import com.akto.dto.graph.SvcToSvcGraph;
+import com.akto.dto.graph.SvcToSvcGraphEdge;
+import com.akto.dto.graph.SvcToSvcGraphNode;
 import com.akto.dto.runtime_filters.RuntimeFilter;
 import com.akto.dto.settings.DataControlSettings;
 import com.akto.dto.test_editor.TestingRunPlayground;
@@ -1767,6 +1770,54 @@ public String findLatestTestingRunResultSummary(){
         return Action.SUCCESS.toUpperCase();
     }
 
+    public List<SvcToSvcGraphEdge> svcToSvcGraphEdges;
+    public List<SvcToSvcGraphNode> svcToSvcGraphNodes;
+
+    public String findSvcToSvcGraphNodes() {
+        try {
+            this.svcToSvcGraphNodes = DbLayer.findSvcToSvcGraphNodes(startTimestamp, endTimestamp, skip, limit);
+        } catch (Exception e) {
+            System.out.println("Error in findSvcToSvcGraphNodes " + e.toString());
+            return Action.ERROR.toUpperCase();
+        }
+
+        return Action.SUCCESS.toUpperCase();
+
+    }
+
+    public String findSvcToSvcGraphEdges() {
+        try {
+            this.svcToSvcGraphEdges = DbLayer.findSvcToSvcGraphEdges(startTimestamp, endTimestamp, skip, limit);
+        } catch (Exception e) {
+            System.out.println("Error in findSvcToSvcGraphEdges " + e.toString());
+            return Action.ERROR.toUpperCase();
+        }
+
+        return Action.SUCCESS.toUpperCase();
+    }
+
+    public String updateSvcToSvcGraphEdges() {
+        try {
+            DbLayer.updateSvcToSvcGraphEdges(this.svcToSvcGraphEdges);
+        } catch (Exception e) {
+            System.out.println("Error in updateSvcToSvcGraphEdges " + e.toString());
+            return Action.ERROR.toUpperCase();
+        }
+
+        return Action.SUCCESS.toUpperCase();
+    }
+
+    public String updateSvcToSvcGraphNodes() {
+        try {
+            DbLayer.updateSvcToSvcGraphNodes(this.svcToSvcGraphNodes);
+        } catch (Exception e) {
+            System.out.println("Error in updateSvcToSvcGraphNodes " + e.toString());
+            return Action.ERROR.toUpperCase();
+        }
+
+        return Action.SUCCESS.toUpperCase();
+    }    
+
     public List<CustomDataTypeMapper> getCustomDataTypes() {
         return customDataTypes;
     }

apps/database-abstractor/src/main/resources/struts.xml

@@ -1233,6 +1233,50 @@
                 <param name="includeProperties">^actionErrors.*</param>
             </result>
         </action>
+        <action name="api/findSvcToSvcGraphEdges" class="com.akto.action.DbAction" method="findSvcToSvcGraphEdges">
+            <interceptor-ref name="json"/>
+            <interceptor-ref name="defaultStack" />
+            <result name="SUCCESS" type="json"/>
+            <result name="ERROR" type="json">
+                <param name="statusCode">422</param>
+                <param name="ignoreHierarchy">false</param>
+                <param name="includeProperties">^actionErrors.*</param>
+            </result>
+        </action>
+
+        <action name="api/findSvcToSvcGraphNodes" class="com.akto.action.DbAction" method="findSvcToSvcGraphNodes">
+            <interceptor-ref name="json"/>
+            <interceptor-ref name="defaultStack" />
+            <result name="SUCCESS" type="json"/>
+            <result name="ERROR" type="json">
+                <param name="statusCode">422</param>
+                <param name="ignoreHierarchy">false</param>
+                <param name="includeProperties">^actionErrors.*</param>
+            </result>
+        </action>
+
+        <action name="api/updateSvcToSvcGraphEdges" class="com.akto.action.DbAction" method="updateSvcToSvcGraphEdges">
+            <interceptor-ref name="json"/>
+            <interceptor-ref name="defaultStack" />
+            <result name="SUCCESS" type="json"/>
+            <result name="ERROR" type="json">
+                <param name="statusCode">422</param>
+                <param name="ignoreHierarchy">false</param>
+                <param name="includeProperties">^actionErrors.*</param>
+            </result>
+        </action>
+
+        <action name="api/updateSvcToSvcGraphNodes" class="com.akto.action.DbAction" method="updateSvcToSvcGraphNodes">
+            <interceptor-ref name="json"/>
+            <interceptor-ref name="defaultStack" />
+            <result name="SUCCESS" type="json"/>
+            <result name="ERROR" type="json">
+                <param name="statusCode">422</param>
+                <param name="ignoreHierarchy">false</param>
+                <param name="includeProperties">^actionErrors.*</param>
+            </result>
+        </action>
+
 
         <action name="api/fetchTestScript" class="com.akto.action.DbAction" method="fetchTestScript">
             <interceptor-ref name="json"/>

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

@@ -32,6 +32,7 @@
 import com.akto.log.LoggerMaker;
 import com.akto.log.LoggerMaker.LogDb;
 import com.akto.metrics.AllMetrics;
+import com.akto.runtime.SvcToSvcGraphManager;
 import com.akto.runtime.utils.Utils;
 import com.akto.data_actor.DataActor;
 import com.akto.data_actor.DataActorFactory;
@@ -65,6 +66,7 @@ public class APICatalogSync {
     public Map<Integer, APICatalog> dbState;
     public Map<Integer, APICatalog> delta;
     public AktoPolicyNew aktoPolicyNew;
+    public SvcToSvcGraphManager svcToSvcGraphManager = null;
     public Map<SensitiveParamInfo, Boolean> sensitiveParamInfoBooleanMap;
     public static boolean mergeAsyncOutside = true;
     public int lastStiFetchTs = 0;
@@ -104,6 +106,7 @@ public APICatalogSync(String userIdentifier, int thresh, boolean fetchAllSTI, bo
         this.mergedUrls = new HashSet<>();
         if (buildFromDb) {
             buildFromDB(false, fetchAllSTI);
+            this.svcToSvcGraphManager = SvcToSvcGraphManager.createFromEdgesAndNodes(dataActor);
         }
     }
 
@@ -230,6 +233,9 @@ public void computeDelta(URLAggregator origAggregator, boolean triggerTemplateGe
             for (HttpResponseParams responseParams: value) {
                 try {
                     aktoPolicyNew.process(responseParams);
+                    if (svcToSvcGraphManager != null) {
+                        svcToSvcGraphManager.processRecord(responseParams.getSvcToSvcGraphParams());
+                    }
                 } catch (Exception e) {
                     e.printStackTrace();
                     throw new RuntimeException(e);
@@ -1565,6 +1571,12 @@ public void syncWithDB(boolean syncImmediately, boolean fetchAllSTI, SyncLimit s
             now = Context.now();
             loggerMaker.infoAndAddToDb("Finished syncing with db at : " + now, LogDb.RUNTIME);
             lastBuildFromDb = now;
+            if (svcToSvcGraphManager != null){
+                svcToSvcGraphManager.updateWithNewDataAndReturnDelta(dataActor);            
+                if (svcToSvcGraphManager.getLastFetchFromDb() < Context.now() - 6 * 60 * 60) {
+                    svcToSvcGraphManager = SvcToSvcGraphManager.createFromEdgesAndNodes(dataActor);
+                }
+            }
         }
     }
 

libs/dao/pom.xml

@@ -182,6 +182,11 @@
             <artifactId>fastjson2</artifactId>
             <version>2.0.51</version>
         </dependency>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <version>1.18.36</version>
+        </dependency>
     </dependencies>
     <build>
         <plugins>

libs/dao/src/main/java/com/akto/dao/graph/SvcToSvcGraphEdgesDao.java

@@ -0,0 +1,22 @@
+package com.akto.dao.graph;
+
+import com.akto.dao.AccountsContextDao;
+import com.akto.dto.graph.SvcToSvcGraphEdge;
+
+public class SvcToSvcGraphEdgesDao extends AccountsContextDao<SvcToSvcGraphEdge> {
+
+    public static final SvcToSvcGraphEdgesDao instance = new SvcToSvcGraphEdgesDao();
+
+    private SvcToSvcGraphEdgesDao() {}
+
+    @Override
+    public String getCollName() {
+        return "svc_to_svc_graph_edges";
+    }
+
+    @Override
+    public Class<SvcToSvcGraphEdge> getClassT() {
+        return SvcToSvcGraphEdge.class;
+    }
+
+}

libs/dao/src/main/java/com/akto/dao/graph/SvcToSvcGraphNodesDao.java

@@ -0,0 +1,22 @@
+package com.akto.dao.graph;
+
+import com.akto.dao.AccountsContextDao;
+import com.akto.dto.graph.SvcToSvcGraphNode;
+
+public class SvcToSvcGraphNodesDao extends AccountsContextDao<SvcToSvcGraphNode> {
+
+    public static final SvcToSvcGraphNodesDao instance = new SvcToSvcGraphNodesDao();
+
+    private SvcToSvcGraphNodesDao() {}
+
+    @Override
+    public String getCollName() {
+        return "svc_to_svc_graph_nodes";
+    }
+
+    @Override
+    public Class<SvcToSvcGraphNode> getClassT() {
+        return SvcToSvcGraphNode.class;
+    }
+
+}

libs/dao/src/main/java/com/akto/dto/HttpResponseParams.java

@@ -2,6 +2,7 @@
 
 
 import com.akto.dao.context.Context;
+import com.akto.dto.graph.SvcToSvcGraphParams;
 
 import java.util.HashMap;
 import java.util.List;
@@ -28,6 +29,8 @@ public enum Source {
     String sourceIP;
     String destIP;
     String direction;
+    SvcToSvcGraphParams svcToSvcGraphParams;
+
     // K8 pod tags in JSON string
     String tags;
 
@@ -37,12 +40,12 @@ public HttpResponseParams(String type, int statusCode, String status, Map<String
                               HttpRequestParams requestParams, int time, String accountId, boolean isPending, Source source, 
                               String orig, String sourceIP) {
         this(type, statusCode, status, headers, payload, requestParams, time, accountId, isPending, source, orig,
-                sourceIP, "", "");
+                sourceIP, "", "", null, null);
     }
 
     public HttpResponseParams(String type, int statusCode, String status, Map<String, List<String>> headers, String payload,
                               HttpRequestParams requestParams, int time, String accountId, boolean isPending, Source source,
-                              String orig, String sourceIP, String destIP, String direction, String tags) {
+                              String orig, String sourceIP, String destIP, String direction, SvcToSvcGraphParams svcToSvcGraphParams, String tags) {
         this.type = type;
         this.statusCode = statusCode;
         this.status = status;
@@ -57,27 +60,10 @@ public HttpResponseParams(String type, int statusCode, String status, Map<String
         this.sourceIP = sourceIP;
         this.destIP = destIP;
         this.direction = direction;
+        this.svcToSvcGraphParams = svcToSvcGraphParams;
         this.tags = tags;
     }
 
-    public HttpResponseParams(String type, int statusCode, String status, Map<String, List<String>> headers, String payload,
-                              HttpRequestParams requestParams, int time, String accountId, boolean isPending, Source source,
-                              String orig, String sourceIP, String destIP, String direction) {
-        this.type = type;
-        this.statusCode = statusCode;
-        this.status = status;
-        this.headers = headers;
-        this.payload = payload;
-        this.requestParams = requestParams;
-        this.time = time;
-        this.accountId = accountId;
-        this.isPending = isPending;
-        this.source = source;
-        this.orig = orig;
-        this.sourceIP = sourceIP;
-        this.destIP = destIP;
-        this.direction = direction;
-    }
 
     public static boolean validHttpResponseCode(int statusCode)  {
         return statusCode >= 200 && (statusCode < 300 || statusCode == 302);
@@ -96,7 +82,11 @@ public HttpResponseParams copy() {
                 this.isPending,
                 this.source,
                 this.orig,
-                this.sourceIP
+                this.sourceIP,
+                this.destIP,
+                this.direction,
+                this.svcToSvcGraphParams,
+                this.tags
         );
     }
 
@@ -185,6 +175,13 @@ public void setRequestParams(HttpRequestParams requestParams) {
         this.requestParams = requestParams;
     }
 
+    public SvcToSvcGraphParams getSvcToSvcGraphParams() {
+        return svcToSvcGraphParams;
+    }
+
+    public void setSvcToSvcGraphParams(SvcToSvcGraphParams svcToSvcGraphParams) {
+        this.svcToSvcGraphParams = svcToSvcGraphParams;
+    }
     public String getTags() {
         return tags;
     }

libs/dao/src/main/java/com/akto/dto/graph/K8sDaemonsetGraphParams.java

@@ -0,0 +1,26 @@
+package com.akto.dto.graph;
+
+import lombok.*;
+
+@Getter
+@Setter
+@NoArgsConstructor
+@AllArgsConstructor
+@ToString
+@EqualsAndHashCode(callSuper = true)
+public class K8sDaemonsetGraphParams extends SvcToSvcGraphParams {
+
+    public static final String DIRECTION_OUTGOING = "2";
+    public static final String DIRECTION_INCOMING = "1";
+
+    String hostInApiRequest;
+    String processId;
+    String socketId;
+    String daemonsetId;
+    String direction;
+
+    @Override
+    public Type getType() {
+        return Type.K8S;
+    }
+}