diff --git a/atsf_agent/internal/agent/runner.go b/atsf_agent/internal/agent/runner.go index e921a389..95276e27 100644 --- a/atsf_agent/internal/agent/runner.go +++ b/atsf_agent/internal/agent/runner.go @@ -321,7 +321,7 @@ func (r *Runner) nodePayload(nodeID string) protocol.NodePayload { profile := observability.BuildProfile(r.Config, r.StateStore) managedOpenRestyMetrics := observability.CollectManagedOpenRestyMetrics(r.Config) metricSnapshot := observability.BuildSnapshot(r.Config, r.StateStore, managedOpenRestyMetrics) - trafficReport := observability.BuildTrafficReport(r.Config, r.StateStore, managedOpenRestyMetrics) + trafficReport, accessLogs := observability.BuildTrafficObservability(r.Config, r.StateStore, managedOpenRestyMetrics) healthEvents := observability.BuildHealthEvents(snapshot) return protocol.NodePayload{ NodeID: nodeID, @@ -336,13 +336,14 @@ func (r *Runner) nodePayload(nodeID string) protocol.NodePayload { Profile: profile, Snapshot: metricSnapshot, TrafficReport: trafficReport, + AccessLogs: accessLogs, HealthEvents: healthEvents, } } func (r *Runner) prepareHeartbeatPayload(nodeID string) (protocol.NodePayload, []int64) { payload := r.nodePayload(nodeID) - if r.ObservabilityBuffer == nil || payload.Snapshot == nil { + if r.ObservabilityBuffer == nil || (payload.Snapshot == nil && payload.TrafficReport == nil && len(payload.AccessLogs) == 0) { return payload, nil } now := time.Now().UTC() @@ -356,6 +357,7 @@ func (r *Runner) prepareHeartbeatPayload(nodeID string) (protocol.NodePayload, [ WindowStartedAtUnix: windowStartedAtUnix, Snapshot: payload.Snapshot, TrafficReport: payload.TrafficReport, + AccessLogs: payload.AccessLogs, QueuedAtUnix: now.Unix(), } if err := r.ObservabilityBuffer.Upsert(record, retainAfterUnix); err != nil { @@ -379,6 +381,7 @@ func (r *Runner) prepareHeartbeatPayload(nodeID string) (protocol.NodePayload, [ WindowStartedAtUnix: item.WindowStartedAtUnix, Snapshot: item.Snapshot, TrafficReport: item.TrafficReport, + AccessLogs: item.AccessLogs, }) ackWindows = append(ackWindows, item.WindowStartedAtUnix) } diff --git a/atsf_agent/internal/agent/runner_test.go b/atsf_agent/internal/agent/runner_test.go index 2e2d4783..f3e0ef1e 100644 --- a/atsf_agent/internal/agent/runner_test.go +++ b/atsf_agent/internal/agent/runner_test.go @@ -311,7 +311,7 @@ func TestRunnerHeartbeatPayloadIncludesObservabilityExtensions(t *testing.T) { } if err := os.WriteFile( filepath.Join(filepath.Dir(runner.Config.RouteConfigPath), "atsflare_access.log"), - []byte("{\"ts\":\""+time.Now().UTC().Format(time.RFC3339)+"\",\"host\":\"edge.example.com\",\"remote_addr\":\"10.0.0.8\",\"status\":200}\n"), + []byte("{\"ts\":\""+time.Now().UTC().Format(time.RFC3339)+"\",\"host\":\"edge.example.com\",\"path\":\"/\",\"remote_addr\":\"10.0.0.8\",\"status\":200}\n"), 0o644, ); err != nil { t.Fatalf("failed to prepare access log: %v", err) @@ -327,6 +327,9 @@ func TestRunnerHeartbeatPayloadIncludesObservabilityExtensions(t *testing.T) { if firstPayload.TrafficReport == nil || firstPayload.TrafficReport.RequestCount != 1 { t.Fatalf("expected first heartbeat payload to include traffic report, got %+v", firstPayload.TrafficReport) } + if len(firstPayload.AccessLogs) != 1 || firstPayload.AccessLogs[0].Path != "/" { + t.Fatalf("expected first heartbeat payload to include access logs, got %+v", firstPayload.AccessLogs) + } if len(firstPayload.HealthEvents) != 2 { t.Fatalf("expected health events for openresty and sync error, got %+v", firstPayload.HealthEvents) } @@ -341,6 +344,9 @@ func TestRunnerHeartbeatPayloadIncludesObservabilityExtensions(t *testing.T) { if secondPayload.TrafficReport != nil { t.Fatalf("expected unchanged traffic window to be omitted on subsequent heartbeat, got %+v", secondPayload.TrafficReport) } + if len(secondPayload.AccessLogs) != 0 { + t.Fatalf("expected unchanged access log delta to be omitted on subsequent heartbeat, got %+v", secondPayload.AccessLogs) + } } func TestRunnerReplaysBufferedObservabilityAfterHeartbeatRecovery(t *testing.T) { @@ -391,7 +397,7 @@ func TestRunnerReplaysBufferedObservabilityAfterHeartbeatRecovery(t *testing.T) } if err := os.WriteFile( filepath.Join(filepath.Dir(runner.Config.RouteConfigPath), "atsflare_access.log"), - []byte("{\"ts\":\""+time.Now().UTC().Format(time.RFC3339)+"\",\"host\":\"edge.example.com\",\"remote_addr\":\"10.0.0.8\",\"status\":200}\n"), + []byte("{\"ts\":\""+time.Now().UTC().Format(time.RFC3339)+"\",\"host\":\"edge.example.com\",\"path\":\"/\",\"remote_addr\":\"10.0.0.8\",\"status\":200}\n"), 0o644, ); err != nil { t.Fatalf("failed to prepare access log: %v", err) @@ -408,6 +414,9 @@ func TestRunnerReplaysBufferedObservabilityAfterHeartbeatRecovery(t *testing.T) if len(secondPayload.BufferedObservability) != 1 { t.Fatalf("expected second heartbeat to replay one buffered observation, got %+v", secondPayload.BufferedObservability) } + if len(secondPayload.BufferedObservability[0].AccessLogs) != 0 { + t.Fatalf("expected seeded buffered observation to keep empty access logs, got %+v", secondPayload.BufferedObservability[0].AccessLogs) + } replayable, err := bufferStore.Replayable(0, 0) if err != nil { diff --git a/atsf_agent/internal/observability/traffic.go b/atsf_agent/internal/observability/traffic.go index 32922d7f..975c3a38 100644 --- a/atsf_agent/internal/observability/traffic.go +++ b/atsf_agent/internal/observability/traffic.go @@ -21,10 +21,11 @@ type accessLogRecord struct { Timestamp string `json:"ts"` Host string `json:"host"` RemoteAddr string `json:"remote_addr"` + Path string `json:"path"` Status int `json:"status"` } -var combinedAccessLogPattern = regexp.MustCompile(`^(\S+)\s+\S+\s+\S+\s+\[([^\]]+)\]\s+"[^"]*"\s+(\d{3})\s+\S+`) +var combinedAccessLogPattern = regexp.MustCompile(`^(\S+)\s+\S+\s+\S+\s+\[([^\]]+)\]\s+"(?:\S+)\s+(\S+)(?:\s+[^"]*)?"\s+(\d{3})\s+\S+`) type trafficAggregate struct { windowStartedAt time.Time @@ -34,16 +35,37 @@ type trafficAggregate struct { statusCodes map[string]int64 topDomains map[string]int64 visitors map[string]struct{} + logs []protocol.NodeAccessLog } func BuildTrafficReport(cfg *config.Config, stateStore *state.Store, managed *managedOpenRestyMetrics) *protocol.NodeTrafficReport { - if managed != nil && managed.TrafficReport != nil { - return managed.TrafficReport - } + report, _ := BuildTrafficObservability(cfg, stateStore, managed) + return report +} + +func BuildTrafficObservability(cfg *config.Config, stateStore *state.Store, managed *managedOpenRestyMetrics) (*protocol.NodeTrafficReport, []protocol.NodeAccessLog) { if cfg == nil || stateStore == nil { - return nil + if managed != nil && managed.TrafficReport != nil { + return managed.TrafficReport, nil + } + return nil, nil } + aggregate := readAccessLogDelta(cfg, stateStore) + accessLogs := []protocol.NodeAccessLog{} + if aggregate != nil { + accessLogs = aggregate.accessLogs() + } + if managed != nil && managed.TrafficReport != nil { + return managed.TrafficReport, accessLogs + } + if aggregate == nil { + return nil, accessLogs + } + return aggregate.report(), accessLogs +} + +func readAccessLogDelta(cfg *config.Config, stateStore *state.Store) *trafficAggregate { snapshot, err := stateStore.Load() if err != nil { return nil @@ -97,7 +119,7 @@ func BuildTrafficReport(cfg *config.Config, stateStore *state.Store, managed *ma snapshot.AccessLogOffset = currentOffset _ = stateStore.Save(snapshot) - return aggregate.report() + return aggregate } func managedAccessLogPath(cfg *config.Config) string { @@ -146,12 +168,20 @@ func (aggregate *trafficAggregate) consume(line []byte) { if remoteAddr := strings.TrimSpace(record.RemoteAddr); remoteAddr != "" { aggregate.visitors[remoteAddr] = struct{}{} } + aggregate.logs = append(aggregate.logs, protocol.NodeAccessLog{ + LoggedAtUnix: record.Timestamp.Unix(), + RemoteAddr: strings.TrimSpace(record.RemoteAddr), + Host: strings.TrimSpace(record.Host), + Path: normalizeAccessLogPath(record.Path), + StatusCode: record.Status, + }) } type parsedAccessLogRecord struct { Timestamp time.Time Host string RemoteAddr string + Path string Status int } @@ -176,26 +206,28 @@ func parseJSONAccessLogRecord(raw string) (parsedAccessLogRecord, bool) { Timestamp: timestamp, Host: strings.TrimSpace(record.Host), RemoteAddr: strings.TrimSpace(record.RemoteAddr), + Path: normalizeAccessLogPath(record.Path), Status: record.Status, }, true } func parseCombinedAccessLogRecord(raw string) (parsedAccessLogRecord, bool) { matches := combinedAccessLogPattern.FindStringSubmatch(raw) - if len(matches) != 4 { + if len(matches) != 5 { return parsedAccessLogRecord{}, false } timestamp, err := parseAccessLogTime(matches[2]) if err != nil { return parsedAccessLogRecord{}, false } - status, err := strconv.Atoi(matches[3]) + status, err := strconv.Atoi(matches[4]) if err != nil { return parsedAccessLogRecord{}, false } return parsedAccessLogRecord{ Timestamp: timestamp, RemoteAddr: strings.TrimSpace(matches[1]), + Path: normalizeAccessLogPath(matches[3]), Status: status, }, true } @@ -217,6 +249,13 @@ func (aggregate *trafficAggregate) report() *protocol.NodeTrafficReport { } } +func (aggregate *trafficAggregate) accessLogs() []protocol.NodeAccessLog { + if aggregate == nil || len(aggregate.logs) == 0 { + return []protocol.NodeAccessLog{} + } + return append([]protocol.NodeAccessLog(nil), aggregate.logs...) +} + func parseAccessLogTime(value string) (time.Time, error) { trimmed := strings.TrimSpace(value) if trimmed == "" { @@ -258,6 +297,20 @@ type trafficCountItem struct { value int64 } +func normalizeAccessLogPath(value string) string { + trimmed := strings.TrimSpace(value) + if trimmed == "" { + return "" + } + if strings.HasPrefix(trimmed, "http://") || strings.HasPrefix(trimmed, "https://") { + return trimmed + } + if strings.HasPrefix(trimmed, "/") { + return trimmed + } + return "/" + trimmed +} + func topCounts(values map[string]int64, limit int) map[string]int64 { return cloneTrafficCounts(values, limit) } diff --git a/atsf_agent/internal/observability/traffic_test.go b/atsf_agent/internal/observability/traffic_test.go index 6cbb1f37..fca91ebc 100644 --- a/atsf_agent/internal/observability/traffic_test.go +++ b/atsf_agent/internal/observability/traffic_test.go @@ -18,9 +18,9 @@ func TestBuildTrafficReportAggregatesManagedAccessLog(t *testing.T) { } logPath := filepath.Join(filepath.Dir(routeConfigPath), "atsflare_access.log") content := []byte( - "{\"ts\":\"2026-03-14T08:00:00Z\",\"host\":\"app.example.com\",\"remote_addr\":\"10.0.0.1\",\"status\":200}\n" + - "{\"ts\":\"2026-03-14T08:00:05Z\",\"host\":\"app.example.com\",\"remote_addr\":\"10.0.0.2\",\"status\":503}\n" + - "{\"ts\":\"2026-03-14T08:00:08Z\",\"host\":\"api.example.com\",\"remote_addr\":\"10.0.0.1\",\"status\":200}\n", + "{\"ts\":\"2026-03-14T08:00:00Z\",\"host\":\"app.example.com\",\"path\":\"/\",\"remote_addr\":\"10.0.0.1\",\"status\":200}\n" + + "{\"ts\":\"2026-03-14T08:00:05Z\",\"host\":\"app.example.com\",\"path\":\"/healthz\",\"remote_addr\":\"10.0.0.2\",\"status\":503}\n" + + "{\"ts\":\"2026-03-14T08:00:08Z\",\"host\":\"api.example.com\",\"path\":\"/api\",\"remote_addr\":\"10.0.0.1\",\"status\":200}\n", ) if err := os.WriteFile(logPath, content, 0o644); err != nil { t.Fatalf("WriteFile failed: %v", err) @@ -62,7 +62,7 @@ func TestBuildTrafficReportResetsOffsetAfterTruncate(t *testing.T) { t.Fatalf("MkdirAll failed: %v", err) } logPath := filepath.Join(filepath.Dir(routeConfigPath), "atsflare_access.log") - if err := os.WriteFile(logPath, []byte("{\"ts\":\"2026-03-14T09:00:00Z\",\"host\":\"app.example.com\",\"remote_addr\":\"10.0.0.3\",\"status\":200}\n"), 0o644); err != nil { + if err := os.WriteFile(logPath, []byte("{\"ts\":\"2026-03-14T09:00:00Z\",\"host\":\"app.example.com\",\"path\":\"/\",\"remote_addr\":\"10.0.0.3\",\"status\":200}\n"), 0o644); err != nil { t.Fatalf("WriteFile failed: %v", err) } @@ -77,6 +77,34 @@ func TestBuildTrafficReportResetsOffsetAfterTruncate(t *testing.T) { } } +func TestBuildTrafficObservabilityReturnsAccessLogs(t *testing.T) { + tempDir := t.TempDir() + routeConfigPath := filepath.Join(tempDir, "conf.d", "atsflare_routes.conf") + if err := os.MkdirAll(filepath.Dir(routeConfigPath), 0o755); err != nil { + t.Fatalf("MkdirAll failed: %v", err) + } + logPath := filepath.Join(filepath.Dir(routeConfigPath), "atsflare_access.log") + content := []byte( + "{\"ts\":\"2026-03-14T08:00:00Z\",\"host\":\"app.example.com\",\"path\":\"/login\",\"remote_addr\":\"10.0.0.1\",\"status\":200}\n" + + "{\"ts\":\"2026-03-14T08:00:05Z\",\"host\":\"api.example.com\",\"path\":\"/v1/ping\",\"remote_addr\":\"10.0.0.2\",\"status\":502}\n", + ) + if err := os.WriteFile(logPath, content, 0o644); err != nil { + t.Fatalf("WriteFile failed: %v", err) + } + + stateStore := state.NewStore(filepath.Join(tempDir, "state.json")) + report, accessLogs := BuildTrafficObservability(&config.Config{RouteConfigPath: routeConfigPath}, stateStore, nil) + if report == nil || report.RequestCount != 2 { + t.Fatalf("expected traffic report, got %+v", report) + } + if len(accessLogs) != 2 { + t.Fatalf("expected access logs, got %+v", accessLogs) + } + if accessLogs[0].Path != "/login" || accessLogs[1].Path != "/v1/ping" { + t.Fatalf("unexpected access log paths: %+v", accessLogs) + } +} + func TestBuildTrafficReportParsesCombinedAccessLog(t *testing.T) { tempDir := t.TempDir() routeConfigPath := filepath.Join(tempDir, "conf.d", "atsflare_routes.conf") diff --git a/atsf_agent/internal/protocol/agent_api.go b/atsf_agent/internal/protocol/agent_api.go index 95ead961..ed4c8957 100644 --- a/atsf_agent/internal/protocol/agent_api.go +++ b/atsf_agent/internal/protocol/agent_api.go @@ -48,6 +48,7 @@ type NodePayload struct { Profile *NodeSystemProfile `json:"profile,omitempty"` Snapshot *NodeMetricSnapshot `json:"snapshot,omitempty"` TrafficReport *NodeTrafficReport `json:"traffic_report,omitempty"` + AccessLogs []NodeAccessLog `json:"access_logs,omitempty"` BufferedObservability []BufferedObservabilityRecord `json:"buffered_observability,omitempty"` HealthEvents []NodeHealthEvent `json:"health_events"` } @@ -93,10 +94,19 @@ type NodeTrafficReport struct { SourceCountries map[string]int64 `json:"source_countries"` } +type NodeAccessLog struct { + LoggedAtUnix int64 `json:"logged_at_unix"` + RemoteAddr string `json:"remote_addr"` + Host string `json:"host"` + Path string `json:"path"` + StatusCode int `json:"status_code"` +} + type BufferedObservabilityRecord struct { WindowStartedAtUnix int64 `json:"window_started_at_unix"` Snapshot *NodeMetricSnapshot `json:"snapshot,omitempty"` TrafficReport *NodeTrafficReport `json:"traffic_report,omitempty"` + AccessLogs []NodeAccessLog `json:"access_logs,omitempty"` } type NodeHealthEvent struct { diff --git a/atsf_agent/internal/state/observability_buffer.go b/atsf_agent/internal/state/observability_buffer.go index 09467fad..d7f7ae72 100644 --- a/atsf_agent/internal/state/observability_buffer.go +++ b/atsf_agent/internal/state/observability_buffer.go @@ -5,6 +5,7 @@ import ( "os" "path/filepath" "sort" + "strconv" "sync" "atsflare-agent/internal/protocol" @@ -16,6 +17,7 @@ type ObservabilityBufferRecord struct { WindowStartedAtUnix int64 `json:"window_started_at_unix"` Snapshot *protocol.NodeMetricSnapshot `json:"snapshot,omitempty"` TrafficReport *protocol.NodeTrafficReport `json:"traffic_report,omitempty"` + AccessLogs []protocol.NodeAccessLog `json:"access_logs,omitempty"` QueuedAtUnix int64 `json:"queued_at_unix"` } @@ -29,7 +31,7 @@ func NewObservabilityBufferStore(path string) *ObservabilityBufferStore { } func (s *ObservabilityBufferStore) Upsert(record ObservabilityBufferRecord, retainAfterUnix int64) error { - if s == nil || record.WindowStartedAtUnix <= 0 || (record.Snapshot == nil && record.TrafficReport == nil) { + if s == nil || record.WindowStartedAtUnix <= 0 || (record.Snapshot == nil && record.TrafficReport == nil && len(record.AccessLogs) == 0) { return nil } s.mu.Lock() @@ -45,7 +47,7 @@ func (s *ObservabilityBufferStore) Upsert(record ObservabilityBufferRecord, reta if records[index].WindowStartedAtUnix != record.WindowStartedAtUnix { continue } - records[index] = record + records[index] = mergeObservabilityBufferRecord(records[index], record) replaced = true break } @@ -58,6 +60,55 @@ func (s *ObservabilityBufferStore) Upsert(record ObservabilityBufferRecord, reta return s.saveUnlocked(records) } +func mergeObservabilityBufferRecord(existing ObservabilityBufferRecord, incoming ObservabilityBufferRecord) ObservabilityBufferRecord { + merged := existing + if incoming.Snapshot != nil { + merged.Snapshot = incoming.Snapshot + } + if incoming.TrafficReport != nil { + merged.TrafficReport = incoming.TrafficReport + } + merged.AccessLogs = mergeAccessLogs(existing.AccessLogs, incoming.AccessLogs) + if incoming.QueuedAtUnix > 0 { + merged.QueuedAtUnix = incoming.QueuedAtUnix + } + return merged +} + +func mergeAccessLogs(existing []protocol.NodeAccessLog, incoming []protocol.NodeAccessLog) []protocol.NodeAccessLog { + if len(existing) == 0 && len(incoming) == 0 { + return nil + } + merged := make([]protocol.NodeAccessLog, 0, len(existing)+len(incoming)) + seen := make(map[string]struct{}, len(existing)+len(incoming)) + appendIfNeeded := func(items []protocol.NodeAccessLog) { + for _, item := range items { + key := accessLogKey(item) + if key == "" { + continue + } + if _, ok := seen[key]; ok { + continue + } + seen[key] = struct{}{} + merged = append(merged, item) + } + } + appendIfNeeded(existing) + appendIfNeeded(incoming) + sort.Slice(merged, func(i int, j int) bool { + if merged[i].LoggedAtUnix == merged[j].LoggedAtUnix { + return accessLogKey(merged[i]) < accessLogKey(merged[j]) + } + return merged[i].LoggedAtUnix < merged[j].LoggedAtUnix + }) + return merged +} + +func accessLogKey(item protocol.NodeAccessLog) string { + return strconv.FormatInt(item.LoggedAtUnix, 10) + "|" + item.RemoteAddr + "|" + item.Host + "|" + item.Path + "|" + strconv.Itoa(item.StatusCode) +} + func (s *ObservabilityBufferStore) Replayable(currentWindowStartedAtUnix int64, retainAfterUnix int64) ([]ObservabilityBufferRecord, error) { if s == nil { return nil, nil diff --git a/atsf_agent/internal/state/observability_buffer_test.go b/atsf_agent/internal/state/observability_buffer_test.go index bc64e49f..4f4e31ed 100644 --- a/atsf_agent/internal/state/observability_buffer_test.go +++ b/atsf_agent/internal/state/observability_buffer_test.go @@ -58,6 +58,36 @@ func TestObservabilityBufferStoreUpsertReplayAndAck(t *testing.T) { } } +func TestObservabilityBufferStoreMergesAccessLogsWithinWindow(t *testing.T) { + store := NewObservabilityBufferStore(filepath.Join(t.TempDir(), "observability-buffer.json")) + + if err := store.Upsert(ObservabilityBufferRecord{ + WindowStartedAtUnix: 1710403200, + AccessLogs: []protocol.NodeAccessLog{ + {LoggedAtUnix: 1710403201, RemoteAddr: "10.0.0.1", Host: "app.example.com", Path: "/a", StatusCode: 200}, + }, + }, 1710403000); err != nil { + t.Fatalf("first upsert failed: %v", err) + } + if err := store.Upsert(ObservabilityBufferRecord{ + WindowStartedAtUnix: 1710403200, + AccessLogs: []protocol.NodeAccessLog{ + {LoggedAtUnix: 1710403201, RemoteAddr: "10.0.0.1", Host: "app.example.com", Path: "/a", StatusCode: 200}, + {LoggedAtUnix: 1710403205, RemoteAddr: "10.0.0.2", Host: "app.example.com", Path: "/b", StatusCode: 502}, + }, + }, 1710403000); err != nil { + t.Fatalf("second upsert failed: %v", err) + } + + records, err := store.Replayable(0, 1710403000) + if err != nil { + t.Fatalf("Replayable failed: %v", err) + } + if len(records) != 1 || len(records[0].AccessLogs) != 2 { + t.Fatalf("expected merged access logs, got %+v", records) + } +} + func TestObservabilityWindowStartedAt(t *testing.T) { if value := ObservabilityWindowStartedAt(nil, &protocol.NodeTrafficReport{WindowStartedAtUnix: 1710403200}); value != 1710403200 { t.Fatalf("unexpected traffic window start: %d", value) diff --git a/atsf_server/common/constants.go b/atsf_server/common/constants.go index 977bc1e3..0310a143 100644 --- a/atsf_server/common/constants.go +++ b/atsf_server/common/constants.go @@ -100,7 +100,7 @@ events { http { include mime.types; default_type application/octet-stream; - log_format atsflare_json escape=json '{"ts":"$time_iso8601","host":"$host","remote_addr":"$remote_addr","status":$status,"request_time":$request_time,"bytes_sent":$body_bytes_sent,"request_length":$request_length}'; + log_format atsflare_json escape=json '{"ts":"$time_iso8601","host":"$host","path":"$request_uri","remote_addr":"$remote_addr","status":$status,"request_time":$request_time,"bytes_sent":$body_bytes_sent,"request_length":$request_length}'; access_log {{OpenRestyAccessLogPath}} atsflare_json; sendfile on; tcp_nopush on; diff --git a/atsf_server/controller/access_log.go b/atsf_server/controller/access_log.go new file mode 100644 index 00000000..ffd6e53d --- /dev/null +++ b/atsf_server/controller/access_log.go @@ -0,0 +1,22 @@ +package controller + +import "atsflare/service" + +import "github.com/gin-gonic/gin" + +// GetAccessLogs godoc +// @Summary List access logs +// @Tags AccessLogs +// @Produce json +// @Security BearerAuth +// @Param node_id query string false "Node ID" +// @Success 200 {object} map[string]interface{} +// @Router /api/access-logs/ [get] +func GetAccessLogs(c *gin.Context) { + logs, err := service.ListAccessLogs(c.Query("node_id")) + if err != nil { + respondFailure(c, err.Error()) + return + } + respondSuccess(c, logs) +} diff --git a/atsf_server/model/main.go b/atsf_server/model/main.go index 200732e0..68e99e97 100644 --- a/atsf_server/model/main.go +++ b/atsf_server/model/main.go @@ -107,6 +107,10 @@ func InitDB() (err error) { if err != nil { return err } + err = db.AutoMigrate(&NodeAccessLog{}) + if err != nil { + return err + } err = db.AutoMigrate(&NodeHealthEvent{}) if err != nil { return err diff --git a/atsf_server/model/node_access_log.go b/atsf_server/model/node_access_log.go new file mode 100644 index 00000000..87086b5c --- /dev/null +++ b/atsf_server/model/node_access_log.go @@ -0,0 +1,30 @@ +package model + +import "time" + +type NodeAccessLog struct { + ID uint `json:"id" gorm:"primaryKey"` + NodeID string `json:"node_id" gorm:"index;size:64;not null"` + LoggedAt time.Time `json:"logged_at" gorm:"index"` + RemoteAddr string `json:"remote_addr" gorm:"size:128"` + Host string `json:"host" gorm:"size:255"` + Path string `json:"path" gorm:"size:2048"` + StatusCode int `json:"status_code"` + RawJSON string `json:"raw_json" gorm:"type:text"` + CreatedAt time.Time `json:"created_at"` +} + +func ListNodeAccessLogs(nodeID string, since time.Time, limit int) (logs []*NodeAccessLog, err error) { + query := DB.Order("logged_at desc, id desc") + if nodeID != "" { + query = query.Where("node_id = ?", nodeID) + } + if !since.IsZero() { + query = query.Where("logged_at >= ?", since) + } + if limit > 0 { + query = query.Limit(limit) + } + err = query.Find(&logs).Error + return logs, err +} diff --git a/atsf_server/router/api-router.go b/atsf_server/router/api-router.go index 5fdc8f53..f561f94f 100644 --- a/atsf_server/router/api-router.go +++ b/atsf_server/router/api-router.go @@ -133,6 +133,11 @@ func SetApiRouter(router *gin.Engine) { { applyLogRoute.GET("/", controller.GetApplyLogs) } + accessLogRoute := apiRouter.Group("/access-logs") + accessLogRoute.Use(middleware.AdminAuth()) + { + accessLogRoute.GET("/", controller.GetAccessLogs) + } agentRoute := apiRouter.Group("/agent") { discoveryRoute := agentRoute.Group("/") diff --git a/atsf_server/service/access_log.go b/atsf_server/service/access_log.go new file mode 100644 index 00000000..2555e5f3 --- /dev/null +++ b/atsf_server/service/access_log.go @@ -0,0 +1,55 @@ +package service + +import ( + "atsflare/model" + "strings" + "time" +) + +const accessLogListLimit = 500 + +type AccessLogView struct { + ID uint `json:"id"` + NodeID string `json:"node_id"` + NodeName string `json:"node_name"` + LoggedAt time.Time `json:"logged_at"` + RemoteAddr string `json:"remote_addr"` + Host string `json:"host"` + Path string `json:"path"` + StatusCode int `json:"status_code"` +} + +func ListAccessLogs(nodeID string) ([]AccessLogView, error) { + logs, err := model.ListNodeAccessLogs(strings.TrimSpace(nodeID), time.Now().Add(-nodeAccessLogRetentionWindow), accessLogListLimit) + if err != nil { + return nil, err + } + nodes, err := model.ListNodes() + if err != nil { + return nil, err + } + nodeNames := make(map[string]string, len(nodes)) + for _, node := range nodes { + if node == nil { + continue + } + nodeNames[node.NodeID] = node.Name + } + views := make([]AccessLogView, 0, len(logs)) + for _, item := range logs { + if item == nil { + continue + } + views = append(views, AccessLogView{ + ID: item.ID, + NodeID: item.NodeID, + NodeName: nodeNames[item.NodeID], + LoggedAt: item.LoggedAt, + RemoteAddr: item.RemoteAddr, + Host: item.Host, + Path: item.Path, + StatusCode: item.StatusCode, + }) + } + return views, nil +} diff --git a/atsf_server/service/agent.go b/atsf_server/service/agent.go index f743ac3c..ee10537b 100644 --- a/atsf_server/service/agent.go +++ b/atsf_server/service/agent.go @@ -36,6 +36,7 @@ type AgentNodePayload struct { Profile *AgentNodeSystemProfile `json:"profile,omitempty"` Snapshot *AgentNodeMetricSnapshot `json:"snapshot,omitempty"` TrafficReport *AgentNodeTrafficReport `json:"traffic_report,omitempty"` + AccessLogs []AgentNodeAccessLog `json:"access_logs,omitempty"` BufferedObservability []AgentBufferedObservabilityRecord `json:"buffered_observability,omitempty"` HealthEvents []AgentNodeHealthEvent `json:"health_events"` } diff --git a/atsf_server/service/node_update_test.go b/atsf_server/service/node_update_test.go index 93d61fe6..635dd9d2 100644 --- a/atsf_server/service/node_update_test.go +++ b/atsf_server/service/node_update_test.go @@ -620,6 +620,22 @@ func TestHeartbeatNodePersistsObservabilityPayload(t *testing.T) { TopDomains: map[string]int64{"example.com": 900}, SourceCountries: map[string]int64{"CN": 700, "US": 200}, }, + AccessLogs: []AgentNodeAccessLog{ + { + LoggedAtUnix: time.Now().Add(-45 * time.Second).Unix(), + RemoteAddr: "203.0.113.10", + Host: "example.com", + Path: "/login", + StatusCode: 200, + }, + { + LoggedAtUnix: time.Now().Add(-40 * time.Second).Unix(), + RemoteAddr: "198.51.100.20", + Host: "api.example.com", + Path: "/v1/ping", + StatusCode: 502, + }, + }, HealthEvents: []AgentNodeHealthEvent{ { EventType: "openresty_unhealthy", @@ -657,6 +673,14 @@ func TestHeartbeatNodePersistsObservabilityPayload(t *testing.T) { t.Fatalf("unexpected request reports: %+v", reports) } + accessLogs, err := model.ListNodeAccessLogs(node.NodeID, time.Time{}, 10) + if err != nil { + t.Fatalf("expected node access logs query to succeed: %v", err) + } + if len(accessLogs) != 2 || accessLogs[0].Path == "" { + t.Fatalf("unexpected access logs: %+v", accessLogs) + } + events, err := model.ListNodeHealthEvents(node.NodeID, true, 10) if err != nil { t.Fatalf("expected node health events query to succeed: %v", err) @@ -724,6 +748,15 @@ func TestHeartbeatNodePersistsBufferedObservabilityPayload(t *testing.T) { TopDomains: map[string]int64{"edge.example.com": 40}, SourceCountries: map[string]int64{"CN": 20}, }, + AccessLogs: []AgentNodeAccessLog{ + { + LoggedAtUnix: now.Add(-110 * time.Second).Unix(), + RemoteAddr: "203.0.113.21", + Host: "edge.example.com", + Path: "/buffered", + StatusCode: 200, + }, + }, }, }, }) @@ -747,6 +780,14 @@ func TestHeartbeatNodePersistsBufferedObservabilityPayload(t *testing.T) { t.Fatalf("expected current and buffered reports, got %+v", reports) } + accessLogs, err := model.ListNodeAccessLogs(node.NodeID, time.Time{}, 10) + if err != nil { + t.Fatalf("expected node access logs query to succeed: %v", err) + } + if len(accessLogs) != 1 || accessLogs[0].Path != "/buffered" { + t.Fatalf("expected buffered access logs to persist, got %+v", accessLogs) + } + _, err = HeartbeatNode(node, AgentNodePayload{ NodeID: node.NodeID, Name: node.Name, diff --git a/atsf_server/service/observability.go b/atsf_server/service/observability.go index 73a7618f..b7cba337 100644 --- a/atsf_server/service/observability.go +++ b/atsf_server/service/observability.go @@ -17,6 +17,7 @@ const ( NodeHealthSeverityInfo = "info" NodeHealthSeverityWarning = "warning" NodeHealthSeverityCritical = "critical" + nodeAccessLogRetentionWindow = 24 * time.Hour ) type AgentNodeSystemProfile struct { @@ -60,10 +61,19 @@ type AgentNodeTrafficReport struct { SourceCountries map[string]int64 `json:"source_countries"` } +type AgentNodeAccessLog struct { + LoggedAtUnix int64 `json:"logged_at_unix"` + RemoteAddr string `json:"remote_addr"` + Host string `json:"host"` + Path string `json:"path"` + StatusCode int `json:"status_code"` +} + type AgentBufferedObservabilityRecord struct { WindowStartedAtUnix int64 `json:"window_started_at_unix"` Snapshot *AgentNodeMetricSnapshot `json:"snapshot,omitempty"` TrafficReport *AgentNodeTrafficReport `json:"traffic_report,omitempty"` + AccessLogs []AgentNodeAccessLog `json:"access_logs,omitempty"` } type AgentNodeHealthEvent struct { @@ -78,7 +88,7 @@ func persistHeartbeatObservability(nodeID string, payload AgentNodePayload, repo if strings.TrimSpace(nodeID) == "" { return } - if payload.Profile == nil && payload.Snapshot == nil && payload.TrafficReport == nil && len(payload.BufferedObservability) == 0 && payload.HealthEvents == nil { + if payload.Profile == nil && payload.Snapshot == nil && payload.TrafficReport == nil && len(payload.AccessLogs) == 0 && len(payload.BufferedObservability) == 0 && payload.HealthEvents == nil { return } @@ -95,6 +105,9 @@ func persistHeartbeatObservability(nodeID string, payload AgentNodePayload, repo if err := persistNodeTrafficReport(tx, nodeID, payload.TrafficReport, reportedAt); err != nil { return err } + if err := persistNodeAccessLogs(tx, nodeID, payload.AccessLogs, reportedAt); err != nil { + return err + } if payload.HealthEvents != nil { if err := reconcileNodeHealthEvents(tx, nodeID, payload.HealthEvents, reportedAt); err != nil { return err @@ -114,6 +127,9 @@ func persistBufferedObservability(tx *gorm.DB, nodeID string, records []AgentBuf if err := persistNodeTrafficReport(tx, nodeID, record.TrafficReport, reportedAt); err != nil { return err } + if err := persistNodeAccessLogs(tx, nodeID, record.AccessLogs, reportedAt); err != nil { + return err + } } return nil } @@ -186,6 +202,35 @@ func persistNodeTrafficReport(tx *gorm.DB, nodeID string, report *AgentNodeTraff return tx.Where("node_id = ? AND window_started_at = ? AND window_ended_at = ?", nodeID, record.WindowStartedAt, record.WindowEndedAt).Assign(record).FirstOrCreate(record).Error } +func persistNodeAccessLogs(tx *gorm.DB, nodeID string, logs []AgentNodeAccessLog, reportedAt time.Time) error { + if len(logs) == 0 { + return nil + } + for _, item := range logs { + record := &model.NodeAccessLog{ + NodeID: nodeID, + LoggedAt: timeFromUnix(item.LoggedAtUnix, reportedAt), + RemoteAddr: strings.TrimSpace(item.RemoteAddr), + Host: strings.TrimSpace(item.Host), + Path: strings.TrimSpace(item.Path), + StatusCode: item.StatusCode, + RawJSON: marshalJSON(item), + } + if err := tx.Where( + "node_id = ? AND logged_at = ? AND remote_addr = ? AND host = ? AND path = ? AND status_code = ?", + nodeID, + record.LoggedAt, + record.RemoteAddr, + record.Host, + record.Path, + record.StatusCode, + ).Assign(record).FirstOrCreate(record).Error; err != nil { + return err + } + } + return tx.Where("node_id = ? AND logged_at < ?", nodeID, reportedAt.Add(-nodeAccessLogRetentionWindow)).Delete(&model.NodeAccessLog{}).Error +} + func reconcileNodeHealthEvents(tx *gorm.DB, nodeID string, events []AgentNodeHealthEvent, reportedAt time.Time) error { activeTypes := make(map[string]AgentNodeHealthEvent, len(events)) for _, event := range events { diff --git a/atsf_server/web/app/(dashboard)/access-log/page.tsx b/atsf_server/web/app/(dashboard)/access-log/page.tsx new file mode 100644 index 00000000..58a232d7 --- /dev/null +++ b/atsf_server/web/app/(dashboard)/access-log/page.tsx @@ -0,0 +1,5 @@ +import { AccessLogsPage } from '@/features/access-logs/components/access-logs-page'; + +export default function AccessLogsRoute() { + return ; +} diff --git a/atsf_server/web/components/layout/dashboard-sidebar.tsx b/atsf_server/web/components/layout/dashboard-sidebar.tsx index 92027b5d..222e7f8a 100644 --- a/atsf_server/web/components/layout/dashboard-sidebar.tsx +++ b/atsf_server/web/components/layout/dashboard-sidebar.tsx @@ -80,6 +80,15 @@ function SidebarIcon({ icon }: { icon: NavigationIconKey }) { ); + case 'log': + return ( + + + + + + + ); case 'performance': return ( diff --git a/atsf_server/web/features/access-logs/api/access-logs.ts b/atsf_server/web/features/access-logs/api/access-logs.ts new file mode 100644 index 00000000..adf8edaf --- /dev/null +++ b/atsf_server/web/features/access-logs/api/access-logs.ts @@ -0,0 +1,11 @@ +import { apiRequest } from '@/lib/api/client'; + +import type { AccessLogItem } from '@/features/access-logs/types'; + +export function getAccessLogs(nodeId?: string) { + const normalizedNodeId = nodeId?.trim(); + const query = normalizedNodeId + ? `?node_id=${encodeURIComponent(normalizedNodeId)}` + : ''; + return apiRequest(`/access-logs/${query}`); +} diff --git a/atsf_server/web/features/access-logs/components/access-logs-page.tsx b/atsf_server/web/features/access-logs/components/access-logs-page.tsx new file mode 100644 index 00000000..dbd45661 --- /dev/null +++ b/atsf_server/web/features/access-logs/components/access-logs-page.tsx @@ -0,0 +1,216 @@ +'use client'; + +import Link from 'next/link'; +import { useMemo, useState } from 'react'; +import { useQuery, useQueryClient } from '@tanstack/react-query'; + +import { EmptyState } from '@/components/feedback/empty-state'; +import { ErrorState } from '@/components/feedback/error-state'; +import { LoadingState } from '@/components/feedback/loading-state'; +import { PageHeader } from '@/components/layout/page-header'; +import { AppCard } from '@/components/ui/app-card'; +import { StatusBadge } from '@/components/ui/status-badge'; +import { getAccessLogs } from '@/features/access-logs/api/access-logs'; +import type { AccessLogItem } from '@/features/access-logs/types'; +import { + PrimaryButton, + ResourceInput, + SecondaryButton, +} from '@/features/shared/components/resource-primitives'; +import { formatDateTime, formatRelativeTime } from '@/lib/utils/date'; + +const accessLogsQueryKey = (nodeId: string) => ['access-logs', nodeId] as const; + +function getErrorMessage(error: unknown) { + return error instanceof Error ? error.message : '请求失败，请稍后重试。'; +} + +function buildSummary(logs: AccessLogItem[]) { + const uniqueIPs = new Set(logs.map((item) => item.remote_addr).filter(Boolean)); + const uniqueNodes = new Set(logs.map((item) => item.node_id).filter(Boolean)); + + return [ + { label: '访问记录', value: logs.length }, + { label: '来源 IP', value: uniqueIPs.size }, + { label: '命中节点', value: uniqueNodes.size }, + { + label: '5xx 响应', + value: logs.filter((item) => item.status_code >= 500).length, + }, + ]; +} + +function getStatusMeta(statusCode: number) { + if (statusCode >= 500) { + return { label: String(statusCode), variant: 'danger' as const }; + } + if (statusCode >= 400) { + return { label: String(statusCode), variant: 'warning' as const }; + } + return { label: String(statusCode), variant: 'success' as const }; +} + +export function AccessLogsPage() { + const queryClient = useQueryClient(); + const [nodeFilterInput, setNodeFilterInput] = useState(''); + const [nodeFilter, setNodeFilter] = useState(''); + + const logsQuery = useQuery({ + queryKey: accessLogsQueryKey(nodeFilter), + queryFn: () => getAccessLogs(nodeFilter), + }); + + const logs = useMemo(() => logsQuery.data ?? [], [logsQuery.data]); + const summary = useMemo(() => buildSummary(logs), [logs]); + + return ( +
+ + 返回节点页 + + } + /> + + +
+ {summary.map((item) => ( +
+

+ {item.label} +

+

+ {item.value} +

+
+ ))} +
+ + + + void queryClient.invalidateQueries({ + queryKey: accessLogsQueryKey(nodeFilter), + }) + } + > + 刷新 + + } + > +
+
+ setNodeFilterInput(event.target.value)} + placeholder="输入 node_id 过滤访问日志" + className="lg:max-w-md" + /> +
+ setNodeFilter(nodeFilterInput.trim())} + > + 筛选 + + { + setNodeFilter(''); + setNodeFilterInput(''); + }} + > + 清空 + +
+
+ + {logsQuery.isLoading ? ( + + ) : logsQuery.isError ? ( + + ) : logs.length === 0 ? ( + + ) : ( +
+ + + + + + + + + + + + + {logs.map((item) => { + const statusMeta = getStatusMeta(item.status_code); + return ( + + + + + + + + + ); + })} + +
时间原 IP访问域名路径节点状态码
+
{formatDateTime(item.logged_at)}
+
+ {formatRelativeTime(item.logged_at)} +
+ 		+ {item.remote_addr || '—'} + + {item.host || '—'} + + {item.path || '—'} + +
{item.node_name || item.node_id}
+
+ {item.node_id} +
+ 		+ +
+
+ )} +
+ +
+ ); +} diff --git a/atsf_server/web/features/access-logs/types.ts b/atsf_server/web/features/access-logs/types.ts new file mode 100644 index 00000000..a3c2b214 --- /dev/null +++ b/atsf_server/web/features/access-logs/types.ts @@ -0,0 +1,10 @@ +export interface AccessLogItem { + id: number; + node_id: string; + node_name: string; + logged_at: string; + remote_addr: string; + host: string; + path: string; + status_code: number; +} diff --git a/atsf_server/web/lib/constants/navigation.ts b/atsf_server/web/lib/constants/navigation.ts index db844958..0ab3f7d8 100644 --- a/atsf_server/web/lib/constants/navigation.ts +++ b/atsf_server/web/lib/constants/navigation.ts @@ -26,6 +26,11 @@ export const dashboardNavigation: NavigationItem[] = [ label: '发布', icon: 'release', }, + { + href: '/access-log', + label: '日志', + icon: 'log', + }, { href: '/performance', diff --git a/atsf_server/web/types/navigation.ts b/atsf_server/web/types/navigation.ts index 34c2224b..b7147358 100644 --- a/atsf_server/web/types/navigation.ts +++ b/atsf_server/web/types/navigation.ts @@ -6,6 +6,7 @@ export type NavigationIconKey = | 'certificate' | 'proxy' | 'release' + | 'log' | 'performance' | 'user' | 'setting'; diff --git a/docs/design.md b/docs/design.md index 56507a49..881e6823 100644 --- a/docs/design.md +++ b/docs/design.md @@ -82,7 +82,7 @@ ATSFlare 当前定位为内部自用的反向代理控制面，不面向外部 * 为了支持世界地图看板，允许节点维护低频地图元数据，如位置名、纬度和经度；这类字段仍属于控制面摘要信息，可保留在 `nodes` * 节点详情页聚焦系统信息、实时资源、网络流量、24 小时趋势和目标版本，不继续堆积低价值静态字段 * 在不引入 Prometheus、ClickHouse、Kafka 等新基础设施前提下完成第六版；数据采集、聚合与查询继续落在现有 Server/SQLite 基线内 -* 原始请求数据不作为长期日志平台对外开放；第六版允许保留受控时间窗口内的明细，用于聚合分析、趋势计算与节点详情辅助排查 +* 原始请求数据不作为长期日志平台对外开放；第六版允许保留受控时间窗口内的明细，用于聚合分析、趋势计算、节点详情辅助排查，以及管理端“日志”页面查看最近时间窗口内的访问记录 * 第六版不做完整 APM、调用链追踪、日志检索平台、任意 SQL 分析接口或自定义报表系统 新增能力超出上述边界时，必须先更新本文档，再进入实现。 @@ -181,6 +181,7 @@ ATSFlare Agent (register / heartbeat / sync / apply / update) * `tls_certificates`：托管证书与私钥 * `managed_domains`：域名资产及默认证书关系 * `node_request_reports`：节点通过 heartbeat 批量上报的请求明细或请求批次 +* `node_access_logs`：节点最近时间窗口内的受控访问明细，仅保留用于管理端排查的必要字段 * `node_metric_snapshots`：节点实时资源、磁盘 IO 与网络流量快照 * `traffic_analytics_rollups`：按时间窗口聚合后的访问指标，用于总览与节点详情看板 * `node_health_events`：节点运行状态变化、阈值异常与配置偏差事件 @@ -203,6 +204,7 @@ ATSFlare Agent (register / heartbeat / sync / apply / update) * 节点请求数据必须随 heartbeat 按批次上报，Server 不主动反向拉取节点日志文件 * 指标看板使用服务端聚合结果，不在前端重复做大规模统计计算 * 节点资源快照与请求明细必须能按时间排序并绑定节点，保证 24 小时趋势与节点详情可回放 +* 管理端日志页只展示受控保留窗口内的必要访问字段，如原始来源 IP、访问域名、路径、命中的节点与响应状态码，不演变为通用日志检索平台 * 原始请求明细、聚合统计与节点基础状态必须在时间窗口上可对齐 * 首页总览的系统状态必须基于统一服务端口径生成，不能由多个历史列表接口在前端临时拼装推导 * 健康事件必须支持“触发中/已恢复”状态，避免首页异常永远累积 diff --git a/docs/development-guidelines.md b/docs/development-guidelines.md index bdc95b37..a1a5c4ed 100644 --- a/docs/development-guidelines.md +++ b/docs/development-guidelines.md @@ -133,6 +133,7 @@ * `tls_certificates` * `managed_domains` * `node_request_reports` +* `node_access_logs` * `node_metric_snapshots` * `traffic_analytics_rollups` * `node_health_events` @@ -154,6 +155,7 @@ * 第六版新增的请求明细、资源快照和聚合统计必须按节点与时间窗口关联 * `node_metric_snapshots` 必须是追加式时间序列快照，不通过覆盖 `nodes` 当前值替代历史 * `traffic_analytics_rollups` 必须区分时间粒度与统计范围，优先存储窗口聚合而不是无限制保留原始逐请求明细 +* `node_access_logs` 仅保留管理端排查所需的受控访问字段与短期保留窗口，不承担全文检索或长期归档职责 * `node_health_events` 必须具备事件类型、严重级别、首次触发时间、最近触发时间和恢复时间，便于首页总览做异常归并 * 访问分析优先复用现有 Server/SQLite 基线，不为第六版引入新的时序数据库或消息队列 * 聚合统计与原始明细的保留策略必须明确，避免无限制累积