[Feature]add metrics delete on write (DataLinkDC#3643)

Co-authored-by: gaoyan1998 <[email protected]>

dinky-admin/src/main/java/org/dinky/context/MetricsContextHolder.java

@@ -60,7 +60,7 @@ public class MetricsContextHolder {
     protected static final MetricsContextHolder instance = new MetricsContextHolder();
 
     private final List<MetricsVO> metricsVOS = new CopyOnWriteArrayList<>();
-    private final AtomicLong lastDumpTime = new AtomicLong(System.currentTimeMillis());
+    private final AtomicLong lastDumpTime = new AtomicLong(0);
 
     static {
         String sql = String.format(

dinky-admin/src/main/java/org/dinky/init/SystemInit.java

@@ -36,7 +36,6 @@
 import org.dinky.function.pool.UdfCodePool;
 import org.dinky.job.ClearJobHistoryTask;
 import org.dinky.job.FlinkJobTask;
-import org.dinky.job.SystemMetricsTask;
 import org.dinky.resource.BaseResourceManager;
 import org.dinky.scheduler.client.ProjectClient;
 import org.dinky.scheduler.exception.SchedulerException;
@@ -54,7 +53,6 @@
 
 import java.util.List;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
 import org.springframework.boot.ApplicationArguments;
@@ -146,23 +144,6 @@ private void initResources() {
      * init task monitor
      */
     private void initDaemon() {
-        SystemConfiguration sysConfig = SystemConfiguration.getInstances();
-
-        // Init system metrics task
-        DaemonTask sysMetricsTask = DaemonTask.build(new DaemonTaskConfig(SystemMetricsTask.TYPE));
-        Configuration<Boolean> metricsSysEnable = sysConfig.getMetricsSysEnable();
-        Configuration<Integer> sysGatherTiming = sysConfig.getMetricsSysGatherTiming();
-        Consumer<Configuration<?>> metricsListener = c -> {
-            c.addChangeEvent(x -> {
-                schedule.removeSchedule(sysMetricsTask.getType());
-                PeriodicTrigger trigger = new PeriodicTrigger(sysGatherTiming.getValue());
-                if (metricsSysEnable.getValue()) schedule.addSchedule(sysMetricsTask, trigger);
-            });
-        };
-        metricsListener.accept(metricsSysEnable);
-        metricsListener.accept(sysGatherTiming);
-        metricsSysEnable.runChangeEvent();
-
         // Init clear job history task
         DaemonTask clearJobHistoryTask = DaemonTask.build(new DaemonTaskConfig(ClearJobHistoryTask.TYPE));
         schedule.addSchedule(clearJobHistoryTask, new PeriodicTrigger(1, TimeUnit.HOURS));

dinky-admin/src/main/java/org/dinky/utils/SqliteUtil.java

@@ -26,6 +26,7 @@
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -34,16 +35,18 @@ public enum SqliteUtil {
     INSTANCE;
 
     private Connection connection;
+    private final AtomicLong lastRecyle = new AtomicLong(0);
 
     static {
         try {
             SqliteUtil.INSTANCE.connect("dinky.db");
+            SqliteUtil.INSTANCE.recyleData();
         } catch (SQLException e) {
             throw new RuntimeException(e);
         }
     }
 
-    public void connect(String dbPath) throws SQLException {
+    private void connect(String dbPath) throws SQLException {
         connection = DriverManager.getConnection("jdbc:sqlite:" + dbPath);
     }
 
@@ -57,11 +60,24 @@ public void createTable(String tableName, String columns) {
         }
     }
 
-    public void write(String tableName, String columns, String values) throws SQLException {
-        String sql = String.format("INSERT INTO %s (%s) VALUES (%s);", tableName, columns, values);
+    public void executeSql(String sql) throws SQLException {
+        Statement pstmt = connection.createStatement();
+        pstmt.executeUpdate(sql);
+        connection.commit();
+    }
 
-        try (PreparedStatement pstmt = connection.prepareStatement(sql)) {
-            pstmt.executeUpdate();
+    public void recyleData() {
+        long now = System.currentTimeMillis();
+        if (now - lastRecyle.get() < 1000 * 60 * 60) {
+            return;
+        }
+        lastRecyle.set(now);
+        try {
+            String sql = "DELETE FROM dinky_metrics WHERE heart_time <= datetime('now', '-7 days')";
+            executeSql(sql);
+            executeSql("VACUUM");
+        } catch (SQLException e) {
+            log.error("Failed to recyle database: " + e.getMessage());
         }
     }
 
@@ -81,6 +97,7 @@ public void write(String tableName, List<String> columns, List<List<String>> val
         } catch (SQLException e) {
             log.error("Failed to write to SQLite: " + e.getMessage());
         }
+        recyleData();
     }
 
     private static String createInsertSql(String tableName, List<String> columns) {