[INLONG-10745][Audit] Unify the range of audit aggregation intervals

inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/config/SqlConstants.java

@@ -54,7 +54,7 @@ public class SqlConstants {
                     ", SUM(count) AS cnt, SUM(size) AS size\n" +
                     "\t\t\t, SUM(delay) AS delay\n" +
                     "\t\tFROM audit_data\n" +
-                    "\t\tWHERE log_ts BETWEEN ? AND ?\n" +
+                    "\t\tWHERE log_ts >= ? AND log_ts < ? \n" +
                     "\t\t\tAND audit_id = ?\n" +
                     "\t\tGROUP BY audit_version, log_ts, inlong_group_id, inlong_stream_id, audit_id, audit_tag\n" +
                     "\t) t_all_version\n" +
@@ -66,7 +66,7 @@ public class SqlConstants {
                     "    ELSE audit_tag\n" +
                     "END AS audit_tag \n" +
                     "\t\t\tFROM audit_data\n" +
-                    "\t\t\tWHERE log_ts BETWEEN ? AND ?\n" +
+                    "\t\t\tWHERE log_ts >= ? AND log_ts < ? \n" +
                     "\t\t\t\tAND audit_id = ?\n" +
                     "\t\t\tGROUP BY log_ts, inlong_group_id, inlong_stream_id, audit_id, audit_tag\n" +
                     "\t\t) t_max_version\n" +
@@ -84,7 +84,7 @@ public class SqlConstants {
             "SELECT ip, sum(count) AS cnt, sum(size) AS size\n" +
                     "\t, sum(delay) AS delay\n" +
                     "FROM audit_data\n" +
-                    "WHERE log_ts BETWEEN ? AND ?\n" +
+                    "WHERE log_ts >= ? AND log_ts < ? \n" +
                     "\tAND inlong_group_id = ? \n" +
                     "\tAND inlong_stream_id =  ? \n" +
                     "\tAND audit_id =  ? \n" +
@@ -100,7 +100,7 @@ public class SqlConstants {
                     "\t, sum(count) AS cnt, sum(size) AS size\n" +
                     "\t, sum(delay) AS delay\n" +
                     "FROM audit_data\n" +
-                    "WHERE log_ts BETWEEN ? AND ? \n" +
+                    "WHERE log_ts >= ? AND log_ts < ? \n" +
                     "\tAND audit_id = ? \n" +
                     "\tAND ip = ? \n" +
                     "GROUP BY inlong_group_id, inlong_stream_id, audit_id, audit_tag";
@@ -120,7 +120,7 @@ public class SqlConstants {
                     "    END AS audit_tag ," +
                     " count, size, delay\n" +
                     "\tFROM audit_data\n" +
-                    "\tWHERE log_ts BETWEEN ? AND ?\n" +
+                    "\tWHERE log_ts >= ? AND log_ts < ? \n" +
                     "\t\tAND inlong_group_id = ?\n" +
                     "\t\tAND inlong_stream_id = ?\n" +
                     "\t\tAND audit_id = ?\n" +
@@ -137,14 +137,14 @@ public class SqlConstants {
                     ", sum(count) AS cnt, sum(size) AS size\n" +
                     ", sum(delay) AS delay\n" +
                     "FROM audit_data_temp\n" +
-                    "WHERE log_ts BETWEEN ?  AND ? \n" +
+                    "WHERE log_ts >= ? AND log_ts < ? \n" +
                     "AND audit_id = ? \n" +
                     "GROUP BY inlong_group_id, inlong_stream_id, audit_id, audit_tag";
 
     public static final String KEY_MYSQL_SOURCE_QUERY_DAY_SQL = "mysql.query.day.sql";
     public static final String DEFAULT_MYSQL_SOURCE_QUERY_DAY_SQL =
             "select log_ts,inlong_group_id,inlong_stream_id,audit_id,audit_tag,count,size,delay " +
-                    "from audit_data_day where log_ts between ? and ? and inlong_group_id=? and inlong_stream_id=? and audit_id =? ";
+                    "from audit_data_day where log_ts >= ? AND log_ts < ? AND inlong_group_id=? AND inlong_stream_id=? AND audit_id =? ";
 
     public static final String KEY_MYSQL_QUERY_AUDIT_ID_SQL = "mysql.query.audit.id.sql";
     public static final String DEFAULT_MYSQL_QUERY_AUDIT_ID_SQL =

inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/source/JdbcSource.java

@@ -140,7 +140,7 @@ public List<StartEndTime> getStatCycleOfMinute(int hoursAgo, int dataCycle) {
             StartEndTime statCycle = new StartEndTime();
             statCycle.setStartTime(dateFormat.format(calendar.getTime()));
 
-            calendar.set(Calendar.MINUTE, minute + dataCycle - 1);
+            calendar.set(Calendar.MINUTE, minute + dataCycle);
             calendar.set(Calendar.SECOND, 0);
             statCycle.setEndTime(dateFormat.format(calendar.getTime()));
             statCycleList.add(statCycle);

inlong-audit/audit-service/src/test/java/org/apache/inlong/audit/source/JdbcSourceTest.java

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.audit.source;
+
+import org.apache.inlong.audit.entities.StartEndTime;
+
+import org.junit.Test;
+
+import java.time.Duration;
+import java.time.LocalTime;
+import java.time.format.DateTimeFormatter;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class JdbcSourceTest {
+
+    @Test
+    public void getStatCycleOfMinute() {
+        JdbcSource jdbcSource = new JdbcSource(null, null);
+
+        List<StartEndTime> startEndTime = jdbcSource.getStatCycleOfMinute(1, 5);
+        assertEquals(12, startEndTime.size());
+
+        DateTimeFormatter formatterHour = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+        LocalTime firstElement = LocalTime.parse(startEndTime.get(0).getStartTime(), formatterHour);
+        LocalTime lastElement = LocalTime.parse(startEndTime.get(11).getEndTime(), formatterHour);
+        Duration duration = Duration.between(firstElement, lastElement);
+        assertEquals(60, duration.toMinutes());
+
+        DateTimeFormatter formatterMinutes = DateTimeFormatter.ofPattern("HH:mm:ss");
+        for (StartEndTime entry : startEndTime) {
+            LocalTime startTime = LocalTime.parse(entry.getStartTime().substring(11), formatterMinutes);
+            boolean check = startTime.getMinute() % 5 == 0 && startTime.getSecond() == 0;
+            assertTrue(check);
+
+            LocalTime endTime = LocalTime.parse(entry.getEndTime().substring(11), formatterMinutes);
+            check = endTime.getMinute() % 5 == 0 && endTime.getSecond() == 0;
+            assertTrue(check);
+        }
+    }
+}

inlong-audit/audit-service/src/test/java/utils/CacheUtilsTest.java → inlong-audit/audit-service/src/test/java/org/apache/inlong/audit/utils/CacheUtilsTest.java

@@ -15,9 +15,7 @@
  * limitations under the License.
  */
 
-package utils;
-
-import org.apache.inlong.audit.utils.CacheUtils;
+package org.apache.inlong.audit.utils;
 
 import org.junit.Test;