mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-10-07 16:16:37 +08:00
feat: add access log functionality and related components
- Introduced NodeAccessLog model and corresponding database migrations. - Implemented access log retrieval in the service layer. - Created API endpoint for accessing logs with appropriate security measures. - Developed frontend components for displaying access logs, including filtering and summary statistics. - Updated observability buffer to include access logs and ensure proper merging and retention. - Enhanced traffic report and observability tests to validate new access log features. - Updated documentation to reflect changes in access log handling and data retention policies.
This commit is contained in:
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user