add testing alerts logic

apps/dashboard/src/main/java/com/akto/action/testing/StartTestAction.java

@@ -2,12 +2,14 @@
 
 import com.akto.action.ExportSampleDataAction;
 import com.akto.action.UserAction;
+import com.akto.dao.TestingAlertsDao;
 import com.akto.dao.context.Context;
 import com.akto.dao.test_editor.YamlTemplateDao;
 import com.akto.dao.testing.sources.TestSourceConfigsDao;
 import com.akto.dao.testing_run_findings.TestingRunIssuesDao;
 import com.akto.dao.testing.*;
 import com.akto.dto.ApiInfo;
+import com.akto.dto.TestingAlerts;
 import com.akto.dto.User;
 import com.akto.dto.ApiToken.Utility;
 import com.akto.dto.CollectionConditions.TestConfigsAdvancedSettings;
@@ -212,6 +214,8 @@ public String startTest() {
             } else {
                 TestingRunDao.instance.insertOne(localTestingRun);
                 testingRunHexId = localTestingRun.getId().toHexString();
+                TestingAlerts testingAlerts = new TestingAlerts(Context.accountId.get(), localTestingRun.getId(), "SCHEDULED", Context.now(), false);
+                TestingAlertsDao.instance.insertOne(testingAlerts);
             }
             this.testIdConfig = 0;
         } else {

apps/dashboard/src/main/java/com/akto/listener/InitializerListener.java

@@ -2290,6 +2290,12 @@ public void accept(Account account) {
                         }
                     }, "context-initializer-secondary");
 
+                    executorService.schedule(new Runnable() {
+                        public void run() {
+                            crons.testingAlertsScheduler();
+                        }
+                    }, 0, TimeUnit.SECONDS);
+
                     crons.trafficAlertsScheduler();
                     crons.insertHistoricalDataJob();
                     if(DashboardMode.isOnPremDeployment()){

apps/dashboard/src/main/java/com/akto/utils/crons/Crons.java

@@ -9,32 +9,44 @@
 
 import com.akto.dao.ApiInfoDao;
 import com.akto.dao.HistoricalDataDao;
+import com.akto.dao.TestingAlertsDao;
 import com.akto.dao.context.Context;
 import com.akto.dto.ApiInfo;
 import com.akto.dto.HistoricalData;
+import com.akto.dto.TestingAlerts;
 import com.akto.listener.InitializerListener;
 import com.akto.log.LoggerMaker;
+import com.akto.log.LoggerMaker.LogDb;
 
-import com.akto.task.Cluster;
 import org.bson.conversions.Bson;
 import org.bson.types.ObjectId;
 
 import com.akto.dao.testing.DeleteTestRunsDao;
+import com.akto.dao.testing.TestingRunDao;
 import com.akto.dao.traffic_metrics.RuntimeMetricsDao;
 import com.akto.dao.traffic_metrics.TrafficAlertsDao;
 import com.akto.dto.Account;
 import com.akto.dto.testing.DeleteTestRuns;
+import com.akto.dto.testing.TestingRun;
 import com.akto.dto.traffic_metrics.RuntimeMetrics;
 import com.akto.dto.traffic_metrics.TrafficAlerts;
 import com.akto.dto.traffic_metrics.TrafficAlerts.ALERT_TYPE;
 import com.akto.util.AccountTask;
 import com.akto.util.enums.GlobalEnums.Severity;
+import com.akto.util.http_util.CoreHTTPClient;
 import com.akto.utils.DeleteTestRunUtils;
 import com.mongodb.BasicDBObject;
 import com.mongodb.client.MongoCursor;
 import com.mongodb.client.model.Accumulators;
 import com.mongodb.client.model.Aggregates;
 import com.mongodb.client.model.Filters;
+import com.mongodb.client.model.FindOneAndUpdateOptions;
+import com.mongodb.client.model.Updates;
+import com.slack.api.Slack;
+import com.slack.api.util.http.SlackHttpClient;
+import com.slack.api.webhook.WebhookResponse;
+
+import okhttp3.OkHttpClient;
 
 
 public class Crons {
@@ -133,6 +145,45 @@ public void accept(Account t) {
         }, 0 , 5, TimeUnit.MINUTES);
     }
 
+    public void testingAlertsScheduler(){
+        scheduler.scheduleAtFixedRate(new Runnable() {
+            public void run(){
+                logger.infoAndAddToDb("testing alerts scheduler triggered", LogDb.DASHBOARD);
+                Bson filters = Filters.and(
+                    Filters.lte("updatedTs", Context.now() - 15 * 60),
+                    Filters.eq("status", "SCHEDULED"),
+                    Filters.eq("alertSent", false)
+                );
+                List<TestingAlerts> testingAlerts = TestingAlertsDao.instance.findAll(filters);
+
+                for (TestingAlerts alert: testingAlerts) {
+                    TestingRun tr = TestingRunDao.instance.findOne(Filters.eq("_id", alert.getTestRunId()));
+                    if (tr != null && tr.getScheduleTimestamp() < Context.now()) {
+                        OkHttpClient httpClient = CoreHTTPClient.client.newBuilder().build();
+                        SlackHttpClient slackHttpClient = new SlackHttpClient(httpClient);
+                        Slack slack = Slack.getInstance(slackHttpClient);
+                        String webhookUrl = "https://hooks.slack.com/triggers/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";
+                        try {
+                            BasicDBObject payload = new BasicDBObject();
+                            payload.put("accountId", alert.getAccountId());
+                            payload.put("testRunId", alert.getTestRunId().toHexString());
+                            logger.infoAndAddToDb("Test alert payload:" + payload, LogDb.DASHBOARD);
+                            WebhookResponse response = slack.send(webhookUrl, payload.toJson());
+                            logger.infoAndAddToDb("Test alert Response: " + response.getBody(), LogDb.DASHBOARD);
+                            Bson updates = Updates.combine(
+                                Updates.set("alertSent", true)
+                            );
+                            TestingAlertsDao.instance.getMCollection().findOneAndUpdate(Filters.eq("testRunId", tr.getId()), updates, new FindOneAndUpdateOptions());
+                        } catch (Exception e) {
+                            e.printStackTrace();
+                            logger.errorAndAddToDb(e, "Error while sending testing alert: " + e.getMessage(), LogDb.DASHBOARD);
+                        }
+                    }
+                }
+            }
+        }, 0 , 5, TimeUnit.MINUTES);
+    }
+
     public static void insertHistoricalData(){
         int currentTime = Context.now();
         Map<Integer, HistoricalData> historicalDataMap = new HashMap<>();

apps/testing/src/main/java/com/akto/testing/Main.java

@@ -290,6 +290,12 @@ public void run() {
                     return;
                 }
 
+                Bson updates = Updates.combine(
+                    Updates.set("updatedTs", Context.now()),
+                    Updates.set("status", State.RUNNING)
+                );
+                TestingAlertsDao.instance.getMCollection().findOneAndUpdate(Filters.eq("testRunId", testingRun.getId()), updates, new FindOneAndUpdateOptions());
+
                 if (testingRun.getState().equals(State.STOPPED)) {
                     loggerMaker.infoAndAddToDb("Testing run stopped");
                     if (trrs != null) {
@@ -470,25 +476,37 @@ public void run() {
                         Updates.set(TestingRun.END_TIMESTAMP, Context.now())
                 );
 
+                Bson alertUpdates = null;
                 if (testingRun.getPeriodInSeconds() > 0 ) {
                     completedUpdate = Updates.combine(
                             Updates.set(TestingRun.STATE, TestingRun.State.SCHEDULED),
                             Updates.set(TestingRun.END_TIMESTAMP, Context.now()),
                             Updates.set(TestingRun.SCHEDULE_TIMESTAMP, testingRun.getScheduleTimestamp() + testingRun.getPeriodInSeconds())
                     );
+                    alertUpdates = Updates.combine(
+                        Updates.set("updatedTs", Context.now()),
+                        Updates.set("status", TestingRun.State.SCHEDULED)
+                    );
                 } else if (testingRun.getPeriodInSeconds() == -1) {
                     completedUpdate = Updates.combine(
                             Updates.set(TestingRun.STATE, TestingRun.State.SCHEDULED),
                             Updates.set(TestingRun.END_TIMESTAMP, Context.now()),
                             Updates.set(TestingRun.SCHEDULE_TIMESTAMP, testingRun.getScheduleTimestamp() + 5 * 60)
                     );
+                    alertUpdates = Updates.combine(
+                        Updates.set("updatedTs", Context.now()),
+                        Updates.set("status", TestingRun.State.SCHEDULED)
+                    );
                 }
 
                 if(GetRunningTestsStatus.getRunningTests().isTestRunning(testingRun.getId())){
                     loggerMaker.infoAndAddToDb("Updating status of running test to Completed.");
                     TestingRunDao.instance.getMCollection().findOneAndUpdate(
                             Filters.eq("_id", testingRun.getId()),  completedUpdate
                     );
+                    if (alertUpdates != null) {
+                        TestingAlertsDao.instance.getMCollection().findOneAndUpdate(Filters.eq("testRunId", testingRun.getId()), alertUpdates, new FindOneAndUpdateOptions());
+                    }
                 }
 
                 if(summaryId != null && testingRun.getTestIdConfig() != 1){

libs/dao/src/main/java/com/akto/DaoInit.java

@@ -407,6 +407,7 @@ public static void createIndices() {
         TrafficAlertsDao.instance.createIndicesIfAbsent();
         RuntimeMetricsDao.instance.createIndicesIfAbsent();
         ApiAuditLogsDao.instance.createIndicesIfAbsent();
+        TestingAlertsDao.instance.createIndicesIfAbsent();
     }
 
 }

libs/dao/src/main/java/com/akto/dao/TestingAlertsDao.java

@@ -0,0 +1,41 @@
+package com.akto.dao;
+
+import com.akto.dao.context.Context;
+import com.akto.dto.TestingAlerts;
+
+public class TestingAlertsDao extends CommonContextDao<TestingAlerts>{
+    public static final TestingAlertsDao instance = new TestingAlertsDao();
+
+    public void createIndicesIfAbsent() {
+
+        boolean exists = false;
+        for (String col: clients[0].getDatabase(Context.accountId.get()+"").listCollectionNames()){
+            if (getCollName().equalsIgnoreCase(col)){
+                exists = true;
+                break;
+            }
+        };
+
+        if (!exists) {
+            clients[0].getDatabase(Context.accountId.get()+"").createCollection(getCollName());
+        }
+
+        String[] fieldNames = {"updatedTs"};
+        MCollection.createIndexIfAbsent(getDBName(), getCollName(), fieldNames, true);
+
+        fieldNames = new String[] {"testRunId"};
+        MCollection.createIndexIfAbsent(getDBName(), getCollName(), fieldNames, true);
+    }
+
+
+    @Override
+    public String getCollName() {
+        return "testing_alerts";
+    }
+
+    @Override
+    public Class<TestingAlerts> getClassT() {
+        return TestingAlerts.class;
+    }
+}
+

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

@@ -0,0 +1,73 @@
+package com.akto.dto;
+
+import org.bson.types.ObjectId;
+
+public class TestingAlerts {
+
+    private ObjectId id;
+    private int accountId;
+    private ObjectId testRunId;
+    private String status;
+    private int updatedTs;
+    private boolean alertSent;
+
+    public TestingAlerts() {
+    }
+
+    public TestingAlerts(int accountId, ObjectId testRunId, String status, int updatedTs, boolean alertSent) {
+        this.accountId = accountId;
+        this.testRunId = testRunId;
+        this.status = status;
+        this.updatedTs = updatedTs;
+        this.alertSent = alertSent;
+    }
+
+    public ObjectId getId() {
+        return id;
+    }
+
+    public void setId(ObjectId id) {
+        this.id = id;
+    }
+
+    public int getAccountId() {
+        return accountId;
+    }
+
+    public void setAccountId(int accountId) {
+        this.accountId = accountId;
+    }
+
+    public ObjectId getTestRunId() {
+        return testRunId;
+    }
+
+    public void setTestRunId(ObjectId testRunId) {
+        this.testRunId = testRunId;
+    }
+
+    public String getStatus() {
+        return status;
+    }
+
+    public void setStatus(String status) {
+        this.status = status;
+    }
+
+    public int getUpdatedTs() {
+        return updatedTs;
+    }
+
+    public void setUpdatedTs(int updatedTs) {
+        this.updatedTs = updatedTs;
+    }
+
+    public boolean isAlertSent() {
+        return alertSent;
+    }
+
+    public void setAlertSent(boolean alertSent) {
+        this.alertSent = alertSent;
+    }
+
+}