scheduled cleaning in javatraces , and add null pointer checking

collector/docker/build-asyncprofiler.sh

@@ -1,5 +1,5 @@
-ASYNC_PROFILER=async-profiler-1.0.3-linux-x64.tar.gz
-KINDLING_JAVA=kindling-java-1.0.3.tar.gz
+ASYNC_PROFILER=async-profiler-1.0.4-linux-x64.tar.gz
+KINDLING_JAVA=kindling-java-1.0.4.tar.gz
 APM_ALL=apm-all-3.1.0.jar
 
 SCRIPT_DIR="$(cd "$(dirname "$SCRIPT_BIN")" > /dev/null 2>&1; pwd -P)"

collector/docker/kindling-collector-config.yml

@@ -51,6 +51,12 @@ analyzers:
     # edge_events_window_size is the size of the duration window that seats the edge events.
     # The unit is second. The greater it is, the more data will be stored.
     edge_events_window_size: 2
+    #java_trace_delete_interval is the interval for cleaning up expired data in javatraces.
+    #The unit is seconds.
+    java_trace_delete_interval: 10
+    #java_trace_expiration_time is the expiration time for data in javatraces.
+    #The unit is seconds.
+    java_trace_expiration_time: 30
   tcpconnectanalyzer:
     channel_size: 10000
     wait_event_second: 10
@@ -242,4 +248,4 @@ observability:
       # Note: DO NOT add the prefix "http://"
       endpoint: 10.10.10.10:8080
     stdout:
-      collect_period: 15s
+      collect_period: 15s

collector/pkg/component/analyzer/cpuanalyzer/config.go

@@ -20,6 +20,13 @@ type Config struct {
 	// EdgeEventsWindowSize is the size of the duration window that seats the edge events.
 	// The unit is seconds. The greater it is, the more data will be stored.
 	EdgeEventsWindowSize int `mapstructure:"edge_events_window_size"`
+	//JavaTraceDeleteInterval is the interval for cleaning up expired data in javatraces.
+	//The unit is seconds.
+	JavaTraceDeleteInterval int `mapstructure:"java_trace_delete_interval"`
+	//JavaTraceExpirationTime is the expiration time for data in javatraces.
+	//The unit is seconds.
+	JavaTraceExpirationTime int `mapstructure:"java_trace_expiration_time"`
+
 }
 
 func NewDefaultConfig() *Config {
@@ -29,5 +36,7 @@ func NewDefaultConfig() *Config {
 		JavaTraceSlowTime:     500,
 		SegmentSize:           40,
 		EdgeEventsWindowSize:  2,
+		JavaTraceDeleteInterval: 10,
+		JavaTraceExpirationTime: 30,
 	}
 }

collector/pkg/component/analyzer/cpuanalyzer/cpu_analyzer.go

@@ -27,17 +27,23 @@ const (
 )
 
 type CpuAnalyzer struct {
-	cfg             *Config
-	cpuPidEvents    map[uint32]map[uint32]*TimeSegments
-	routineSize     *atomic.Int32
-	lock            sync.RWMutex
-	telemetry       *component.TelemetryTools
-	tidExpiredQueue *tidDeleteQueue
-	javaTraces      map[string]*TransactionIdEvent
-	nextConsumers   []consumer.Consumer
-	metadata        *kubernetes.K8sMetaDataCache
+	cfg              *Config
+	cpuPidEvents     map[uint32]map[uint32]*TimeSegments
+	routineSize      *atomic.Int32
+	lock             sync.RWMutex
+	telemetry        *component.TelemetryTools
+	tidExpiredQueue  *tidDeleteQueue
+	javaTraces       map[JavaTracesKey]*TransactionIdEvent
+	nextConsumers    []consumer.Consumer
+	metadata         *kubernetes.K8sMetaDataCache
+	cleanerTicker    *time.Ticker
+	stopProfileChan  chan struct{}
+}
 
-	stopProfileChan chan struct{}
+type JavaTracesKey struct{
+	TraceId     	string 
+	PidString   	string 
+	StartTime   	time.Time 
 }
 
 func (ca *CpuAnalyzer) Type() analyzer.Type {
@@ -59,20 +65,34 @@ func NewCpuAnalyzer(cfg interface{}, telemetry *component.TelemetryTools, consum
 	}
 	ca.cpuPidEvents = make(map[uint32]map[uint32]*TimeSegments, 100000)
 	ca.tidExpiredQueue = newTidDeleteQueue()
-	ca.javaTraces = make(map[string]*TransactionIdEvent, 100000)
+	ca.javaTraces = make(map[JavaTracesKey]*TransactionIdEvent, 100000)
 	newSelfMetrics(telemetry.MeterProvider, ca)
 	return ca
 }
 
 func (ca *CpuAnalyzer) Start() error {
-	// Disable receiving and sending the profiling data by default.
+	ca.cleanerTicker = time.NewTicker(time.Duration(ca.cfg.JavaTraceDeleteInterval) * time.Second)
+	go func() {
+		for range ca.cleanerTicker.C {
+			ca.lock.Lock()
+			now := time.Now()
+			for key:= range ca.javaTraces {
+				if now.Sub(key.StartTime) > time.Duration(ca.cfg.JavaTraceExpirationTime)*time.Second {
+					delete(ca.javaTraces, key)
+					fmt.Print("Expired data has been released,pid = " + key.PidString)
+				}
+			}
+			ca.lock.Unlock()
+		}
+	}()
 	return nil
 }
 
 func (ca *CpuAnalyzer) Shutdown() error {
 	if enableProfile {
 		_ = ca.StopProfile()
 	}
+	ca.cleanerTicker.Stop()
 	return nil
 }
 
@@ -116,10 +136,19 @@ func (ca *CpuAnalyzer) ConsumeTransactionIdEvent(event *model.KindlingEvent) {
 }
 
 func (ca *CpuAnalyzer) analyzerJavaTraceTime(ev *TransactionIdEvent) {
+	javatracekey := &JavaTracesKey{
+		TraceId: ev.TraceId,
+		PidString: ev.PidString,
+		StartTime: time.Now(),
+	}
 	if ev.IsEntry == 1 {
-		ca.javaTraces[ev.TraceId+ev.PidString] = ev
+		ca.javaTraces[*javatracekey] = ev
 	} else {
-		oldEvent := ca.javaTraces[ev.TraceId+ev.PidString]
+		oldEvent,ok := ca.javaTraces[*javatracekey]
+		if(!ok){
+			ca.telemetry.Logger.Warnf("No javaTraces traceid=%d, pid=%s", javatracekey.TraceId,javatracekey.PidString)
+			return
+		}
 		pid, _ := strconv.ParseInt(ev.PidString, 10, 64)
 		spendTime := ev.Timestamp - oldEvent.Timestamp
 		contentKey := oldEvent.Url

collector/pkg/component/analyzer/cpuanalyzer/cpu_analyzer_test.go

@@ -0,0 +1,55 @@
+package cpuanalyzer
+
+import (
+	"math/rand"
+	"strconv"
+	"testing"
+	"time"
+)
+
+func TestStart(t *testing.T) {
+	keys := make([]JavaTracesKey, 10)
+	for i := 0; i < 10; i++ {
+		offset := time.Duration(rand.Int63n(int64(10 * time.Second)))
+		keys[i] = JavaTracesKey{
+			TraceId:   strconv.Itoa(rand.Intn(1000000)),
+			PidString: strconv.Itoa(rand.Intn(1000000)),
+			StartTime: time.Now().Add(offset - 5*time.Second),
+		}
+	}
+	tevent := &TransactionIdEvent{
+		TraceId: "0",
+		PidString: "1",
+	}
+	javaTraces := make(map[JavaTracesKey]*TransactionIdEvent)
+	for _, key := range keys {
+		javaTraces[key] = tevent
+	}
+	config:= NewDefaultConfig()
+	ca := &CpuAnalyzer{
+		javaTraces: javaTraces,
+		cfg:           config,
+	}
+	ca.cleanerTicker = time.NewTicker(time.Duration(ca.cfg.JavaTraceDeleteInterval) * time.Second)
+	go func() {
+		for range ca.cleanerTicker.C {
+			ca.lock.Lock()
+			now := time.Now()
+			for key:= range ca.javaTraces {
+				if now.Sub(key.StartTime) > time.Duration(ca.cfg.JavaTraceExpirationTime)*time.Second {
+					delete(ca.javaTraces, key)
+					t.Log("已删除pid="+
+					key.PidString+ "，当前时间："+
+					time.Now().Truncate(time.Second).Format("15:04:05")+
+					"，map剩余数量:"+strconv.Itoa(len(javaTraces)))
+				}
+			}
+			ca.lock.Unlock()
+		}
+	}()
+	time.Sleep(10*time.Minute)
+}
+
+func Test(t *testing.T){
+
+}

deploy/agent/kindling-collector-config.yml

@@ -51,6 +51,12 @@ analyzers:
     # edge_events_window_size is the size of the duration window that seats the edge events.
     # The unit is second. The greater it is, the more data will be stored.
     edge_events_window_size: 2
+    #java_trace_delete_interval is the interval for cleaning up expired data in javatraces.
+    #The unit is seconds.
+    java_trace_delete_interval: 10
+    #java_trace_expiration_time is the expiration time for data in javatraces.
+    #The unit is seconds.
+    java_trace_expiration_time: 30
   tcpconnectanalyzer:
     channel_size: 10000
     wait_event_second: 10
@@ -242,4 +248,4 @@ observability:
       # Note: DO NOT add the prefix "http://"
       endpoint: 10.10.10.10:8080
     stdout:
-      collect_period: 15s
+      collect_period: 15s