diff --git a/internal/apps/openflare/observability/access_log_logics.go b/internal/apps/openflare/observability/access_log_logics.go index e2e8bbff..98c78dfa 100644 --- a/internal/apps/openflare/observability/access_log_logics.go +++ b/internal/apps/openflare/observability/access_log_logics.go @@ -21,11 +21,15 @@ const ( defaultIPTrendBucketMinute = 30 maxIPTrendHours = 168 nodeAccessLogRetentionDays = 90 + defaultAccessLogQueryDays = 7 accessLogFieldRemoteAddr = "remote_addr" accessLogFieldRequestCount = "request_count" ) -var nodeAccessLogRetentionWindow = nodeAccessLogRetentionDays * 24 * time.Hour +var ( + nodeAccessLogRetentionWindow = nodeAccessLogRetentionDays * 24 * time.Hour + defaultAccessLogQueryWindow = defaultAccessLogQueryDays * 24 * time.Hour +) // AccessLogQuery filters access log list queries. type AccessLogQuery struct { @@ -346,7 +350,7 @@ func ListFoldedAccessLogIPs(ctx context.Context, input FoldedAccessLogIPQuery) ( // ListAccessLogIPSummaries returns paginated IP summaries. func ListAccessLogIPSummaries(ctx context.Context, input AccessLogIPSummaryQuery) (*AccessLogIPSummaryList, error) { normalized := normalizeAccessLogIPSummaryQuery(input) - since := time.Now().UTC().Add(-nodeAccessLogRetentionWindow) + since := defaultAccessLogSince() recentSince := time.Now().UTC().Add(-3 * time.Hour) query := model.OpenFlareAccessLogIPSummaryQuery{ NodeID: strings.TrimSpace(normalized.NodeID), @@ -453,7 +457,7 @@ func buildModelAccessLogQuery(input AccessLogQuery) model.OpenFlareAccessLogQuer RemoteAddr: strings.TrimSpace(input.RemoteAddr), Host: strings.TrimSpace(input.Host), Path: strings.TrimSpace(input.Path), - Since: time.Now().UTC().Add(-nodeAccessLogRetentionWindow), + Since: defaultAccessLogSince(), Page: input.Page, PageSize: input.PageSize, SortBy: input.SortBy, @@ -461,6 +465,10 @@ func buildModelAccessLogQuery(input AccessLogQuery) model.OpenFlareAccessLogQuer } } +func defaultAccessLogSince() time.Time { + return time.Now().UTC().Add(-defaultAccessLogQueryWindow) +} + func listNodeNameMap(ctx context.Context, logs []*model.OpenFlareAccessLog) (map[string]string, error) { nodeIDs := make([]string, 0, len(logs)) seen := make(map[string]struct{}, len(logs)) diff --git a/internal/apps/openflare/waf/ip_group_sync.go b/internal/apps/openflare/waf/ip_group_sync.go index d23dbbc7..4d5effb9 100644 --- a/internal/apps/openflare/waf/ip_group_sync.go +++ b/internal/apps/openflare/waf/ip_group_sync.go @@ -234,15 +234,15 @@ func evaluateParsedIPGroupAutoConfig(ctx context.Context, config ipGroupAutoConf } programs = append(programs, program) } - logs, err := model.ListOpenFlareAccessLogsForWAFIPGroup(ctx, model.OpenFlareAccessLogQuery{ + aggregates, err := model.ListOpenFlareAccessLogWAFIPAggregates(ctx, model.OpenFlareAccessLogQuery{ Since: now.Add(-time.Duration(config.LookbackMinutes) * time.Minute), Until: now, }) if err != nil { return nil, err } - accumulators := make(map[string]*ipGroupAutoAccumulator) - for _, item := range logs { + accumulators := make(map[string]*ipGroupAutoAccumulator, len(aggregates)) + for _, item := range aggregates { if item == nil { continue } @@ -250,30 +250,23 @@ func evaluateParsedIPGroupAutoConfig(ctx context.Context, config ipGroupAutoConf if !ok { continue } - acc := accumulators[ip] - if acc == nil { - acc = &ipGroupAutoAccumulator{ - ip: ip, - statusCounts: make(map[int]int), - } - accumulators[ip] = acc + lastSeen := time.Time{} + if item.LastSeenEpoch > 0 { + lastSeen = time.Unix(item.LastSeenEpoch, 0).UTC() } - acc.requestCount++ - acc.statusCounts[item.StatusCode]++ - if item.StatusCode == http.StatusNotFound { - acc.status404Count++ + statusCounts := make(map[int]int, len(item.StatusCounts)) + for code, count := range item.StatusCounts { + statusCounts[code] = count } - if item.StatusCode >= 400 && item.StatusCode < 500 { - acc.clientErrorCount++ - } - if item.StatusCode >= http.StatusInternalServerError { - acc.serverErrorCount++ - } - if hostIsIPLiteral(item.Host) { - acc.ipHostCount++ - } - if item.LoggedAt.After(acc.lastSeen) { - acc.lastSeen = item.LoggedAt + accumulators[ip] = &ipGroupAutoAccumulator{ + ip: ip, + requestCount: item.RequestCount, + status404Count: item.Status404Count, + ipHostCount: item.IPHostCount, + clientErrorCount: item.ClientErrorCount, + serverErrorCount: item.ServerErrorCount, + lastSeen: lastSeen, + statusCounts: statusCounts, } } matched := make([]string, 0) diff --git a/internal/model/openflare_access_log.go b/internal/model/openflare_access_log.go index 08fe99cf..54220fd5 100644 --- a/internal/model/openflare_access_log.go +++ b/internal/model/openflare_access_log.go @@ -24,6 +24,8 @@ type openFlareAccessLogBucketAggregateRow struct { SuccessCount int64 `gorm:"column:success_count"` ClientErrorCount int64 `gorm:"column:client_error_count"` ServerErrorCount int64 `gorm:"column:server_error_count"` + UniqueIPCount int64 `gorm:"column:unique_ip_count"` + UniqueHostCount int64 `gorm:"column:unique_host_count"` } type openFlareAccessLogBucketDimensionRow struct { @@ -52,9 +54,45 @@ type openFlareAccessLogIPTrendRow struct { RequestCount int64 `gorm:"column:request_count"` } -// ListOpenFlareAccessLogsForWAFIPGroup lists access logs in a time window for automatic IP group rules. -func ListOpenFlareAccessLogsForWAFIPGroup(ctx context.Context, query OpenFlareAccessLogQuery) ([]*OpenFlareAccessLog, error) { - return ListOpenFlareAccessLogs(ctx, query) +type openFlareAccessLogWAFIPAggregateRow struct { + RemoteAddr string + RequestCount int64 + Status404Count int64 + ClientErrorCount int64 + ServerErrorCount int64 + IPHostCount int64 + LastSeenEpoch int64 + StatusCounts map[int]int64 +} + +// ListOpenFlareAccessLogWAFIPAggregates returns per-IP aggregates for WAF automatic rules. +func ListOpenFlareAccessLogWAFIPAggregates(ctx context.Context, query OpenFlareAccessLogQuery) ([]*OpenFlareAccessLogWAFIPAggregate, error) { + rows, err := currentAccessLogStore().WAFIPAggregates(ctx, query) + if err != nil { + return nil, err + } + result := make([]*OpenFlareAccessLogWAFIPAggregate, 0, len(rows)) + for _, row := range rows { + remoteAddr := strings.TrimSpace(row.RemoteAddr) + if remoteAddr == "" { + continue + } + statusCounts := make(map[int]int, len(row.StatusCounts)) + for code, count := range row.StatusCounts { + statusCounts[code] = int(count) + } + result = append(result, &OpenFlareAccessLogWAFIPAggregate{ + RemoteAddr: remoteAddr, + RequestCount: int(row.RequestCount), + Status404Count: int(row.Status404Count), + ClientErrorCount: int(row.ClientErrorCount), + ServerErrorCount: int(row.ServerErrorCount), + IPHostCount: int(row.IPHostCount), + LastSeenEpoch: row.LastSeenEpoch, + StatusCounts: statusCounts, + }) + } + return result, nil } // InsertOpenFlareAccessLogsBatch inserts access log rows into ClickHouse. @@ -79,24 +117,17 @@ func ListOpenFlareAccessLogRegionCounts(ctx context.Context, nodeID string, sinc // ListOpenFlareAccessLogBuckets lists folded access log buckets. func ListOpenFlareAccessLogBuckets(ctx context.Context, query OpenFlareAccessLogBucketQuery) ([]*OpenFlareAccessLogBucketRow, error) { - rows, err := buildOpenFlareAccessLogBucketRows(ctx, query) - if err != nil { - return nil, err - } - start, end := openFlareAccessLogPaginateBounds(len(rows), query.Page, query.PageSize) - if start >= len(rows) { - return []*OpenFlareAccessLogBucketRow{}, nil - } - return rows[start:end], nil + return buildOpenFlareAccessLogBucketRows(ctx, query) } // CountOpenFlareAccessLogBuckets counts folded access log buckets. func CountOpenFlareAccessLogBuckets(ctx context.Context, query OpenFlareAccessLogBucketQuery) (int64, error) { - rows, err := buildOpenFlareAccessLogBucketRows(ctx, query) - if err != nil { - return 0, err + filter := openFlareAccessLogQueryFromBucket(query) + bucketSeconds := int64(query.FoldMinutes * secondsPerMinute) + if bucketSeconds <= 0 { + bucketSeconds = 180 } - return int64(len(rows)), nil + return currentAccessLogStore().CountBuckets(ctx, filter, bucketSeconds) } // ListOpenFlareAccessLogBucketIPs lists folded IP rows for a bucket window. @@ -123,24 +154,13 @@ func CountOpenFlareAccessLogBucketIPs(ctx context.Context, query OpenFlareAccess // ListOpenFlareAccessLogIPSummaries lists IP summaries. func ListOpenFlareAccessLogIPSummaries(ctx context.Context, query OpenFlareAccessLogIPSummaryQuery, recentSince time.Time) ([]*OpenFlareAccessLogIPSummaryRow, error) { - rows, err := buildOpenFlareAccessLogIPSummaryRows(ctx, query, recentSince) - if err != nil { - return nil, err - } - start, end := openFlareAccessLogPaginateBounds(len(rows), query.Page, query.PageSize) - if start >= len(rows) { - return []*OpenFlareAccessLogIPSummaryRow{}, nil - } - return rows[start:end], nil + return buildOpenFlareAccessLogIPSummaryRows(ctx, query, recentSince) } // CountOpenFlareAccessLogIPSummaries counts IP summaries. func CountOpenFlareAccessLogIPSummaries(ctx context.Context, query OpenFlareAccessLogIPSummaryQuery) (int64, error) { - rows, err := buildOpenFlareAccessLogIPSummaryRows(ctx, query, time.Time{}) - if err != nil { - return 0, err - } - return int64(len(rows)), nil + filter := openFlareAccessLogQueryFromIPSummary(query) + return currentAccessLogStore().CountIPSummaries(ctx, filter) } // ListOpenFlareAccessLogIPTrend lists IP trend points. @@ -195,75 +215,22 @@ func buildOpenFlareAccessLogBucketRows(ctx context.Context, query OpenFlareAcces bucketSeconds = 180 } - type bucketAccumulator struct { - requestCount int64 - uniqueIPs map[string]struct{} - uniqueHosts map[string]struct{} - successCount int64 - clientErrorCount int64 - serverErrorCount int64 - } - accumulators := make(map[int64]*bucketAccumulator) - partials, err := currentAccessLogStore().BucketAggregates(ctx, filter, bucketSeconds) if err != nil { return nil, err } + rows := make([]*OpenFlareAccessLogBucketRow, 0, len(partials)) for _, partial := range partials { - accumulator := accumulators[partial.BucketEpoch] - if accumulator == nil { - accumulator = &bucketAccumulator{ - uniqueIPs: make(map[string]struct{}), - uniqueHosts: make(map[string]struct{}), - } - accumulators[partial.BucketEpoch] = accumulator - } - accumulator.requestCount += partial.RequestCount - accumulator.successCount += partial.SuccessCount - accumulator.clientErrorCount += partial.ClientErrorCount - accumulator.serverErrorCount += partial.ServerErrorCount - } - - for _, column := range []string{columnRemoteAddr, columnHost} { - dimensions, err := currentAccessLogStore().BucketDimensions(ctx, filter, column, bucketSeconds) - if err != nil { - return nil, err - } - for _, item := range dimensions { - accumulator := accumulators[item.BucketEpoch] - if accumulator == nil { - accumulator = &bucketAccumulator{ - uniqueIPs: make(map[string]struct{}), - uniqueHosts: make(map[string]struct{}), - } - accumulators[item.BucketEpoch] = accumulator - } - trimmed := strings.TrimSpace(item.Value) - if trimmed == "" { - continue - } - switch column { - case columnRemoteAddr: - accumulator.uniqueIPs[trimmed] = struct{}{} - case columnHost: - accumulator.uniqueHosts[trimmed] = struct{}{} - } - } - } - - rows := make([]*OpenFlareAccessLogBucketRow, 0, len(accumulators)) - for bucketEpoch, accumulator := range accumulators { rows = append(rows, &OpenFlareAccessLogBucketRow{ - BucketEpoch: bucketEpoch, - RequestCount: accumulator.requestCount, - UniqueIPCount: int64(len(accumulator.uniqueIPs)), - UniqueHostCount: int64(len(accumulator.uniqueHosts)), - SuccessCount: accumulator.successCount, - ClientErrorCount: accumulator.clientErrorCount, - ServerErrorCount: accumulator.serverErrorCount, + BucketEpoch: partial.BucketEpoch, + RequestCount: partial.RequestCount, + UniqueIPCount: partial.UniqueIPCount, + UniqueHostCount: partial.UniqueHostCount, + SuccessCount: partial.SuccessCount, + ClientErrorCount: partial.ClientErrorCount, + ServerErrorCount: partial.ServerErrorCount, }) } - sortOpenFlareAccessLogBucketRows(rows, query.SortBy, query.SortOrder) return rows, nil } @@ -293,12 +260,7 @@ func buildOpenFlareAccessLogBucketIPRows(ctx context.Context, query OpenFlareAcc } func buildOpenFlareAccessLogIPSummaryRows(ctx context.Context, query OpenFlareAccessLogIPSummaryQuery, recentSince time.Time) ([]*OpenFlareAccessLogIPSummaryRow, error) { - filter := OpenFlareAccessLogQuery{ - NodeID: query.NodeID, - RemoteAddr: query.RemoteAddr, - Host: query.Host, - Since: query.Since, - } + filter := openFlareAccessLogQueryFromIPSummary(query) partials, err := currentAccessLogStore().IPSummaries(ctx, filter, recentSince) if err != nil { return nil, err @@ -316,7 +278,6 @@ func buildOpenFlareAccessLogIPSummaryRows(ctx context.Context, query OpenFlareAc LastSeenEpoch: partial.LastSeenEpoch, }) } - sortOpenFlareAccessLogIPSummaryRows(rows, query.SortBy, query.SortOrder) return rows, nil } @@ -350,6 +311,23 @@ func openFlareAccessLogQueryFromBucket(query OpenFlareAccessLogBucketQuery) Open Host: query.Host, Path: query.Path, Since: query.Since, + Page: query.Page, + PageSize: query.PageSize, + SortBy: query.SortBy, + SortOrder: query.SortOrder, + } +} + +func openFlareAccessLogQueryFromIPSummary(query OpenFlareAccessLogIPSummaryQuery) OpenFlareAccessLogQuery { + return OpenFlareAccessLogQuery{ + NodeID: query.NodeID, + RemoteAddr: query.RemoteAddr, + Host: query.Host, + Since: query.Since, + Page: query.Page, + PageSize: query.PageSize, + SortBy: query.SortBy, + SortOrder: query.SortOrder, } } diff --git a/internal/model/openflare_access_log_store.go b/internal/model/openflare_access_log_store.go index 9c5f34b1..a69ba5e3 100644 --- a/internal/model/openflare_access_log_store.go +++ b/internal/model/openflare_access_log_store.go @@ -19,9 +19,12 @@ type accessLogStore interface { Count(ctx context.Context, query OpenFlareAccessLogQuery) (int64, int64, error) RegionCounts(ctx context.Context, nodeID string, since time.Time, limit int) ([]*OpenFlareAccessLogRegionCount, error) BucketAggregates(ctx context.Context, filter OpenFlareAccessLogQuery, bucketSeconds int64) ([]openFlareAccessLogBucketAggregateRow, error) + CountBuckets(ctx context.Context, filter OpenFlareAccessLogQuery, bucketSeconds int64) (int64, error) BucketDimensions(ctx context.Context, filter OpenFlareAccessLogQuery, column string, bucketSeconds int64) ([]openFlareAccessLogBucketDimensionRow, error) IPAggregates(ctx context.Context, filter OpenFlareAccessLogQuery, exactRemoteAddr bool) ([]openFlareAccessLogIPAggregateRow, error) + WAFIPAggregates(ctx context.Context, filter OpenFlareAccessLogQuery) ([]openFlareAccessLogWAFIPAggregateRow, error) IPSummaries(ctx context.Context, filter OpenFlareAccessLogQuery, recentSince time.Time) ([]openFlareAccessLogIPSummaryRow, error) + CountIPSummaries(ctx context.Context, filter OpenFlareAccessLogQuery) (int64, error) IPTrend(ctx context.Context, filter OpenFlareAccessLogQuery, bucketSeconds int64) ([]openFlareAccessLogIPTrendRow, error) DeleteAll(ctx context.Context) (int64, error) DeleteBefore(ctx context.Context, cutoff time.Time) (int64, error) @@ -116,11 +119,17 @@ func (clickhouseAccessLogStore) BucketAggregates(ctx context.Context, filter Ope SuccessCount: row.SuccessCount, ClientErrorCount: row.ClientErrorCount, ServerErrorCount: row.ServerErrorCount, + UniqueIPCount: row.UniqueIPCount, + UniqueHostCount: row.UniqueHostCount, } } return result, nil } +func (clickhouseAccessLogStore) CountBuckets(ctx context.Context, filter OpenFlareAccessLogQuery, bucketSeconds int64) (int64, error) { + return analyticsrepo.CountBucketAggregatesNodeAccessLogs(ctx, toNodeAccessLogFilter(filter), bucketSeconds) +} + func (clickhouseAccessLogStore) BucketDimensions(ctx context.Context, filter OpenFlareAccessLogQuery, column string, bucketSeconds int64) ([]openFlareAccessLogBucketDimensionRow, error) { rows, err := analyticsrepo.BucketDimensionsNodeAccessLogs(ctx, toNodeAccessLogFilter(filter), column, bucketSeconds) if err != nil { @@ -172,6 +181,31 @@ func (clickhouseAccessLogStore) IPSummaries(ctx context.Context, filter OpenFlar return result, nil } +func (clickhouseAccessLogStore) CountIPSummaries(ctx context.Context, filter OpenFlareAccessLogQuery) (int64, error) { + return analyticsrepo.CountIPSummaryNodeAccessLogs(ctx, toNodeAccessLogFilter(filter)) +} + +func (clickhouseAccessLogStore) WAFIPAggregates(ctx context.Context, filter OpenFlareAccessLogQuery) ([]openFlareAccessLogWAFIPAggregateRow, error) { + rows, err := analyticsrepo.IPAggregatesForWAFNodeAccessLogs(ctx, toNodeAccessLogFilter(filter)) + if err != nil { + return nil, err + } + result := make([]openFlareAccessLogWAFIPAggregateRow, len(rows)) + for index, row := range rows { + result[index] = openFlareAccessLogWAFIPAggregateRow{ + RemoteAddr: row.RemoteAddr, + RequestCount: row.RequestCount, + Status404Count: row.Status404Count, + ClientErrorCount: row.ClientErrorCount, + ServerErrorCount: row.ServerErrorCount, + IPHostCount: row.IPHostCount, + LastSeenEpoch: row.LastSeenEpoch, + StatusCounts: row.StatusCounts, + } + } + return result, nil +} + func (clickhouseAccessLogStore) IPTrend(ctx context.Context, filter OpenFlareAccessLogQuery, bucketSeconds int64) ([]openFlareAccessLogIPTrendRow, error) { rows, err := analyticsrepo.IPTrendNodeAccessLogs(ctx, toNodeAccessLogFilter(filter), bucketSeconds) if err != nil { diff --git a/internal/model/openflare_access_log_store_memory.go b/internal/model/openflare_access_log_store_memory.go index 728a9367..a707c342 100644 --- a/internal/model/openflare_access_log_store_memory.go +++ b/internal/model/openflare_access_log_store_memory.go @@ -5,6 +5,8 @@ package model import ( "context" + "net" + "net/netip" "sort" "strings" "sync" @@ -99,12 +101,21 @@ func (s *memoryAccessLogStore) BucketAggregates(_ context.Context, filter OpenFl s.mu.RLock() defer s.mu.RUnlock() rows := s.filterRecords(filter) - aggregates := make(map[int64]*openFlareAccessLogBucketAggregateRow) + type bucketAccumulator struct { + openFlareAccessLogBucketAggregateRow + uniqueIPs map[string]struct{} + uniqueHosts map[string]struct{} + } + aggregates := make(map[int64]*bucketAccumulator) for _, row := range rows { bucketEpoch := memoryAccessLogBucketEpoch(row.LoggedAt, bucketSeconds) item := aggregates[bucketEpoch] if item == nil { - item = &openFlareAccessLogBucketAggregateRow{BucketEpoch: bucketEpoch} + item = &bucketAccumulator{ + openFlareAccessLogBucketAggregateRow: openFlareAccessLogBucketAggregateRow{BucketEpoch: bucketEpoch}, + uniqueIPs: make(map[string]struct{}), + uniqueHosts: make(map[string]struct{}), + } aggregates[bucketEpoch] = item } item.RequestCount++ @@ -116,14 +127,61 @@ func (s *memoryAccessLogStore) BucketAggregates(_ context.Context, filter OpenFl default: item.ServerErrorCount++ } + if remoteAddr := strings.TrimSpace(row.RemoteAddr); remoteAddr != "" { + item.uniqueIPs[remoteAddr] = struct{}{} + } + if host := strings.TrimSpace(row.Host); host != "" { + item.uniqueHosts[host] = struct{}{} + } } result := make([]openFlareAccessLogBucketAggregateRow, 0, len(aggregates)) for _, item := range aggregates { - result = append(result, *item) + item.UniqueIPCount = int64(len(item.uniqueIPs)) + item.UniqueHostCount = int64(len(item.uniqueHosts)) + result = append(result, item.openFlareAccessLogBucketAggregateRow) + } + bucketRows := make([]*OpenFlareAccessLogBucketRow, len(result)) + for index := range result { + bucketRows[index] = &OpenFlareAccessLogBucketRow{ + BucketEpoch: result[index].BucketEpoch, + RequestCount: result[index].RequestCount, + UniqueIPCount: result[index].UniqueIPCount, + UniqueHostCount: result[index].UniqueHostCount, + SuccessCount: result[index].SuccessCount, + ClientErrorCount: result[index].ClientErrorCount, + ServerErrorCount: result[index].ServerErrorCount, + } + } + sortOpenFlareAccessLogBucketRows(bucketRows, filter.SortBy, filter.SortOrder) + for index := range result { + result[index] = openFlareAccessLogBucketAggregateRow{ + BucketEpoch: bucketRows[index].BucketEpoch, + RequestCount: bucketRows[index].RequestCount, + UniqueIPCount: bucketRows[index].UniqueIPCount, + UniqueHostCount: bucketRows[index].UniqueHostCount, + SuccessCount: bucketRows[index].SuccessCount, + ClientErrorCount: bucketRows[index].ClientErrorCount, + ServerErrorCount: bucketRows[index].ServerErrorCount, + } + } + if filter.PageSize > 0 { + start, end := openFlareAccessLogPaginateBounds(len(result), filter.Page, filter.PageSize) + return result[start:end], nil } return result, nil } +func (s *memoryAccessLogStore) CountBuckets(_ context.Context, filter OpenFlareAccessLogQuery, bucketSeconds int64) (int64, error) { + s.mu.RLock() + defer s.mu.RUnlock() + rows := s.filterRecords(filter) + seen := make(map[int64]struct{}) + for _, row := range rows { + seen[memoryAccessLogBucketEpoch(row.LoggedAt, bucketSeconds)] = struct{}{} + } + return int64(len(seen)), nil +} + func (s *memoryAccessLogStore) BucketDimensions(_ context.Context, filter OpenFlareAccessLogQuery, column string, bucketSeconds int64) ([]openFlareAccessLogBucketDimensionRow, error) { s.mu.RLock() defer s.mu.RUnlock() @@ -222,9 +280,91 @@ func (s *memoryAccessLogStore) IPSummaries(_ context.Context, filter OpenFlareAc item.LastSeenEpoch = epoch } } - result := make([]openFlareAccessLogIPSummaryRow, 0, len(aggregates)) + summaryRows := make([]*OpenFlareAccessLogIPSummaryRow, 0, len(aggregates)) for _, item := range aggregates { - result = append(result, *item) + summaryRows = append(summaryRows, &OpenFlareAccessLogIPSummaryRow{ + RemoteAddr: item.RemoteAddr, + TotalRequests: item.TotalRequests, + RecentRequests: item.RecentRequests, + LastSeenEpoch: item.LastSeenEpoch, + }) + } + sortOpenFlareAccessLogIPSummaryRows(summaryRows, filter.SortBy, filter.SortOrder) + if filter.PageSize > 0 { + start, end := openFlareAccessLogPaginateBounds(len(summaryRows), filter.Page, filter.PageSize) + summaryRows = summaryRows[start:end] + } + result := make([]openFlareAccessLogIPSummaryRow, len(summaryRows)) + for index, item := range summaryRows { + result[index] = openFlareAccessLogIPSummaryRow{ + RemoteAddr: item.RemoteAddr, + TotalRequests: item.TotalRequests, + RecentRequests: item.RecentRequests, + LastSeenEpoch: item.LastSeenEpoch, + } + } + return result, nil +} + +func (s *memoryAccessLogStore) CountIPSummaries(_ context.Context, filter OpenFlareAccessLogQuery) (int64, error) { + s.mu.RLock() + defer s.mu.RUnlock() + rows := s.filterRecords(filter) + seen := make(map[string]struct{}) + for _, row := range rows { + remoteAddr := strings.TrimSpace(row.RemoteAddr) + if remoteAddr == "" { + continue + } + seen[remoteAddr] = struct{}{} + } + return int64(len(seen)), nil +} + +func (s *memoryAccessLogStore) WAFIPAggregates(_ context.Context, filter OpenFlareAccessLogQuery) ([]openFlareAccessLogWAFIPAggregateRow, error) { + s.mu.RLock() + defer s.mu.RUnlock() + rows := s.filterRecords(filter) + aggregates := make(map[string]*openFlareAccessLogWAFIPAggregateRow) + order := make([]string, 0) + for _, row := range rows { + remoteAddr := strings.TrimSpace(row.RemoteAddr) + if remoteAddr == "" { + continue + } + item := aggregates[remoteAddr] + if item == nil { + item = &openFlareAccessLogWAFIPAggregateRow{ + RemoteAddr: remoteAddr, + StatusCounts: make(map[int]int64), + } + aggregates[remoteAddr] = item + order = append(order, remoteAddr) + } + item.RequestCount++ + item.StatusCounts[row.StatusCode]++ + if row.StatusCode == 404 { + item.Status404Count++ + } + if row.StatusCode >= 400 && row.StatusCode < 500 { + item.ClientErrorCount++ + } + if row.StatusCode >= 500 { + item.ServerErrorCount++ + } + if memoryAccessLogHostIsIPLiteral(row.Host) { + item.IPHostCount++ + } + epoch := row.LoggedAt.UTC().Unix() + if epoch > item.LastSeenEpoch { + item.LastSeenEpoch = epoch + } + } + result := make([]openFlareAccessLogWAFIPAggregateRow, 0, len(order)) + for _, remoteAddr := range order { + if item := aggregates[remoteAddr]; item != nil { + result = append(result, *item) + } } return result, nil } @@ -324,6 +464,19 @@ func memoryAccessLogMatches(row *OpenFlareAccessLog, query OpenFlareAccessLogQue return true } +func memoryAccessLogHostIsIPLiteral(value string) bool { + host := strings.TrimSpace(value) + if host == "" { + return false + } + if parsedHost, _, err := net.SplitHostPort(host); err == nil { + host = parsedHost + } + host = strings.Trim(host, "[]") + _, err := netip.ParseAddr(host) + return err == nil +} + func memoryAccessLogBucketEpoch(loggedAt time.Time, bucketSeconds int64) int64 { if bucketSeconds <= 0 { bucketSeconds = 180 diff --git a/internal/model/openflare_observability.go b/internal/model/openflare_observability.go index 2eeb526f..d47b0099 100644 --- a/internal/model/openflare_observability.go +++ b/internal/model/openflare_observability.go @@ -277,6 +277,18 @@ type OpenFlareAccessLogIPTrendRow struct { RequestCount int64 `json:"request_count"` } +// OpenFlareAccessLogWAFIPAggregate is a per-IP aggregate row for WAF automatic rules. +type OpenFlareAccessLogWAFIPAggregate struct { + RemoteAddr string + RequestCount int + Status404Count int + ClientErrorCount int + ServerErrorCount int + IPHostCount int + LastSeenEpoch int64 + StatusCounts map[int]int +} + func isMissingTableError(err error) bool { if err == nil { return false diff --git a/internal/repository/analytics/access_log_stats.go b/internal/repository/analytics/access_log_stats.go index 42fce01c..7187717e 100644 --- a/internal/repository/analytics/access_log_stats.go +++ b/internal/repository/analytics/access_log_stats.go @@ -99,6 +99,8 @@ func GetBrowserDistribution(ctx context.Context, startTime time.Time) ([]Browser FROM %s WHERE created_at >= ? GROUP BY user_agent + ORDER BY count DESC + LIMIT 100 `, tableName) type uaRow struct { diff --git a/internal/repository/analytics/node_access_log.go b/internal/repository/analytics/node_access_log.go index c5ca2d2c..005f94c6 100644 --- a/internal/repository/analytics/node_access_log.go +++ b/internal/repository/analytics/node_access_log.go @@ -87,23 +87,16 @@ func CountNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter) (int64 clause, args := buildNodeAccessLogFilterClause(filter) tableName := nodeAccessLogTableName() - var totalRecords uint64 - countSQL := fmt.Sprintf("SELECT count() FROM %s WHERE %s", tableName, clause) - if err := conn.QueryRow(ctx, countSQL, args...).Scan(&totalRecords); err != nil { + countSQL := fmt.Sprintf(` +SELECT + count() AS total_records, + uniqExactIf(trim(remote_addr), trim(remote_addr) != '') AS total_ips +FROM %s +WHERE %s`, tableName, clause) + var totalRecords, totalIPs uint64 + if err := conn.QueryRow(ctx, countSQL, args...).Scan(&totalRecords, &totalIPs); err != nil { return 0, 0, fmt.Errorf("count node access logs: %w", err) } - - ipSQL := fmt.Sprintf(` -SELECT count() FROM ( - SELECT trim(remote_addr) AS trimmed_remote_addr - FROM %s - WHERE %s AND trim(remote_addr) != '' - GROUP BY trimmed_remote_addr -)`, tableName, clause) - var totalIPs uint64 - if err := conn.QueryRow(ctx, ipSQL, args...).Scan(&totalIPs); err != nil { - return 0, 0, fmt.Errorf("count node access log ips: %w", err) - } return safeInt64Count(totalRecords), safeInt64Count(totalIPs), nil } diff --git a/internal/repository/analytics/node_access_log_filter.go b/internal/repository/analytics/node_access_log_filter.go index a59b8b19..e89d3441 100644 --- a/internal/repository/analytics/node_access_log_filter.go +++ b/internal/repository/analytics/node_access_log_filter.go @@ -102,6 +102,43 @@ func nodeAccessLogEpochExpr() string { return "toInt64(toUnixTimestamp(logged_at))" } +func nodeAccessLogHostIsIPLiteralExpr() string { + return `( + toIPv4OrNull(trim(if(position(trim(host), ':') > 0 AND NOT startsWith(trim(host), '['), splitByChar(':', trim(host))[1], replaceRegexpAll(trim(host), '\\[|\\]', '')))) IS NOT NULL + OR toIPv6OrNull(trim(if(position(trim(host), ':') > 0 AND NOT startsWith(trim(host), '['), splitByChar(':', trim(host))[1], replaceRegexpAll(trim(host), '\\[|\\]', '')))) IS NOT NULL + )` +} + +func nodeAccessLogBucketOrderClause(sortBy string, sortOrder string) string { + direction := "DESC" + if normalizeNodeAccessLogSortOrder(sortOrder) == "asc" { + direction = "ASC" + } + switch strings.TrimSpace(sortBy) { + case "request_count": + return "request_count " + direction + ", bucket_epoch DESC" + default: + return "bucket_epoch " + direction + } +} + +func nodeAccessLogIPSummaryOrderClause(sortBy string, sortOrder string) string { + direction := "DESC" + if normalizeNodeAccessLogSortOrder(sortOrder) == "asc" { + direction = "ASC" + } + column := "total_requests" + switch strings.TrimSpace(sortBy) { + case "recent_requests": + column = "recent_requests" + case "last_seen_at": + column = "last_seen_epoch" + case "remote_addr": + column = "trimmed_remote_addr" + } + return column + " " + direction + ", last_seen_epoch DESC, trimmed_remote_addr ASC" +} + func nodeAccessLogTableName() string { return "of_node_access_logs" } diff --git a/internal/repository/analytics/node_access_log_stats.go b/internal/repository/analytics/node_access_log_stats.go index 34d79d8e..d7444318 100644 --- a/internal/repository/analytics/node_access_log_stats.go +++ b/internal/repository/analytics/node_access_log_stats.go @@ -17,6 +17,20 @@ type NodeAccessLogBucketAggregate struct { SuccessCount int64 ClientErrorCount int64 ServerErrorCount int64 + UniqueIPCount int64 + UniqueHostCount int64 +} + +// NodeAccessLogWAFIPAggregate is a per-IP aggregate row for WAF automatic rules. +type NodeAccessLogWAFIPAggregate struct { + RemoteAddr string + RequestCount int64 + Status404Count int64 + ClientErrorCount int64 + ServerErrorCount int64 + IPHostCount int64 + LastSeenEpoch int64 + StatusCounts map[int]int64 } // NodeAccessLogBucketDimension is a bucket dimension value. @@ -49,7 +63,7 @@ type NodeAccessLogIPTrend struct { RequestCount int64 } -// BucketAggregatesNodeAccessLogs returns folded bucket aggregates. +// BucketAggregatesNodeAccessLogs returns folded bucket aggregates with unique IP/host counts. func BucketAggregatesNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter, bucketSeconds int64) ([]NodeAccessLogBucketAggregate, error) { conn, err := nodeAccessLogConn() if err != nil { @@ -64,10 +78,20 @@ SELECT count() AS request_count, countIf(status_code < 400) AS success_count, countIf(status_code >= 400 AND status_code < 500) AS client_error_count, - countIf(status_code >= 500) AS server_error_count + countIf(status_code >= 500) AS server_error_count, + uniqExactIf(trim(remote_addr), trim(remote_addr) != '') AS unique_ip_count, + uniqExactIf(trim(host), trim(host) != '') AS unique_host_count FROM %s WHERE %s -GROUP BY bucket_epoch`, bucketExpr, tableName, clause) +GROUP BY bucket_epoch +ORDER BY %s`, bucketExpr, tableName, clause, nodeAccessLogBucketOrderClause(filter.SortBy, filter.SortOrder)) + if filter.PageSize > 0 { + if filter.Page < 0 { + filter.Page = 0 + } + sql += clickHouseLimitOffsetClause + args = append(args, filter.PageSize, filter.Page*filter.PageSize) + } rows, err := conn.Query(ctx, sql, args...) if err != nil { return nil, fmt.Errorf("bucket aggregates node access logs: %w", err) @@ -77,10 +101,10 @@ GROUP BY bucket_epoch`, bucketExpr, tableName, clause) var result []NodeAccessLogBucketAggregate for rows.Next() { var ( - bucketEpoch int64 - requestCount, successCount, clientErrorCount, serverErrorCount uint64 + bucketEpoch int64 + requestCount, successCount, clientErrorCount, serverErrorCount, uniqueIPCount, uniqueHostCount uint64 ) - if err := rows.Scan(&bucketEpoch, &requestCount, &successCount, &clientErrorCount, &serverErrorCount); err != nil { + if err := rows.Scan(&bucketEpoch, &requestCount, &successCount, &clientErrorCount, &serverErrorCount, &uniqueIPCount, &uniqueHostCount); err != nil { return nil, fmt.Errorf("scan bucket aggregate row: %w", err) } result = append(result, NodeAccessLogBucketAggregate{ @@ -89,11 +113,36 @@ GROUP BY bucket_epoch`, bucketExpr, tableName, clause) SuccessCount: safeInt64Count(successCount), ClientErrorCount: safeInt64Count(clientErrorCount), ServerErrorCount: safeInt64Count(serverErrorCount), + UniqueIPCount: safeInt64Count(uniqueIPCount), + UniqueHostCount: safeInt64Count(uniqueHostCount), }) } return result, nil } +// CountBucketAggregatesNodeAccessLogs returns the number of folded buckets matching filter. +func CountBucketAggregatesNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter, bucketSeconds int64) (int64, error) { + conn, err := nodeAccessLogConn() + if err != nil { + return 0, err + } + clause, args := buildNodeAccessLogFilterClause(filter) + bucketExpr := nodeAccessLogBucketEpochExpr(bucketSeconds) + tableName := nodeAccessLogTableName() + sql := fmt.Sprintf(` +SELECT count() FROM ( + SELECT 1 + FROM %s + WHERE %s + GROUP BY %s +)`, tableName, clause, bucketExpr) + var totalBuckets uint64 + if err := conn.QueryRow(ctx, sql, args...).Scan(&totalBuckets); err != nil { + return 0, fmt.Errorf("count bucket aggregates node access logs: %w", err) + } + return safeInt64Count(totalBuckets), nil +} + // BucketDimensionsNodeAccessLogs returns bucket dimension values. func BucketDimensionsNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter, column string, bucketSeconds int64) ([]NodeAccessLogBucketDimension, error) { conn, err := nodeAccessLogConn() @@ -185,7 +234,7 @@ GROUP BY trimmed_remote_addr`, lastSeenExpr, tableName, queryClause) return result, nil } -// IPSummariesNodeAccessLogs returns IP summary rows. +// IPSummariesNodeAccessLogs returns paginated IP summary rows. func IPSummariesNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter, recentSince time.Time) ([]NodeAccessLogIPSummary, error) { conn, err := nodeAccessLogConn() if err != nil { @@ -209,7 +258,15 @@ SELECT max(%s) AS last_seen_epoch FROM %s WHERE %s AND trim(remote_addr) != '' -GROUP BY trimmed_remote_addr`, recentClause, lastSeenExpr, tableName, clause) +GROUP BY trimmed_remote_addr +ORDER BY %s`, recentClause, lastSeenExpr, tableName, clause, nodeAccessLogIPSummaryOrderClause(filter.SortBy, filter.SortOrder)) + if filter.PageSize > 0 { + if filter.Page < 0 { + filter.Page = 0 + } + sql += clickHouseLimitOffsetClause + queryArgs = append(queryArgs, filter.PageSize, filter.Page*filter.PageSize) + } rows, err := conn.Query(ctx, sql, queryArgs...) if err != nil { return nil, fmt.Errorf("ip summaries node access logs: %w", err) @@ -236,6 +293,140 @@ GROUP BY trimmed_remote_addr`, recentClause, lastSeenExpr, tableName, clause) return result, nil } +// CountIPSummaryNodeAccessLogs returns the number of distinct IPs matching filter. +func CountIPSummaryNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter) (int64, error) { + conn, err := nodeAccessLogConn() + if err != nil { + return 0, err + } + clause, args := buildNodeAccessLogFilterClause(filter) + tableName := nodeAccessLogTableName() + sql := fmt.Sprintf(` +SELECT count() FROM ( + SELECT 1 + FROM %s + WHERE %s AND trim(remote_addr) != '' + GROUP BY trim(remote_addr) +)`, tableName, clause) + var totalIPs uint64 + if err := conn.QueryRow(ctx, sql, args...).Scan(&totalIPs); err != nil { + return 0, fmt.Errorf("count ip summary node access logs: %w", err) + } + return safeInt64Count(totalIPs), nil +} + +// IPAggregatesForWAFNodeAccessLogs returns per-IP aggregates for WAF automatic rules. +func IPAggregatesForWAFNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter) ([]NodeAccessLogWAFIPAggregate, error) { + conn, err := nodeAccessLogConn() + if err != nil { + return nil, err + } + clause, args := buildNodeAccessLogFilterClause(filter) + lastSeenExpr := nodeAccessLogEpochExpr() + hostIsIPExpr := nodeAccessLogHostIsIPLiteralExpr() + tableName := nodeAccessLogTableName() + sql := fmt.Sprintf(` +SELECT + trim(remote_addr) AS trimmed_remote_addr, + count() AS request_count, + countIf(status_code = 404) AS status_404_count, + countIf(status_code >= 400 AND status_code < 500) AS client_error_count, + countIf(status_code >= 500) AS server_error_count, + countIf(%s) AS ip_host_count, + max(%s) AS last_seen_epoch +FROM %s +WHERE %s AND trim(remote_addr) != '' +GROUP BY trimmed_remote_addr`, hostIsIPExpr, lastSeenExpr, tableName, clause) + rows, err := conn.Query(ctx, sql, args...) + if err != nil { + return nil, fmt.Errorf("ip aggregates for waf node access logs: %w", err) + } + defer func() { _ = rows.Close() }() + + aggregates := make(map[string]*NodeAccessLogWAFIPAggregate) + order := make([]string, 0) + for rows.Next() { + var ( + remoteAddr string + lastSeenEpoch int64 + requestCount, status404Count, clientErrorCount, serverErrorCount, ipHostCount uint64 + ) + if err := rows.Scan(&remoteAddr, &requestCount, &status404Count, &clientErrorCount, &serverErrorCount, &ipHostCount, &lastSeenEpoch); err != nil { + return nil, fmt.Errorf("scan waf ip aggregate row: %w", err) + } + remoteAddr = strings.TrimSpace(remoteAddr) + if remoteAddr == "" { + continue + } + aggregates[remoteAddr] = &NodeAccessLogWAFIPAggregate{ + RemoteAddr: remoteAddr, + RequestCount: safeInt64Count(requestCount), + Status404Count: safeInt64Count(status404Count), + ClientErrorCount: safeInt64Count(clientErrorCount), + ServerErrorCount: safeInt64Count(serverErrorCount), + IPHostCount: safeInt64Count(ipHostCount), + LastSeenEpoch: lastSeenEpoch, + StatusCounts: make(map[int]int64), + } + order = append(order, remoteAddr) + } + if err := mergeWAFIPStatusCodeCounts(ctx, filter, aggregates); err != nil { + return nil, err + } + result := make([]NodeAccessLogWAFIPAggregate, 0, len(order)) + for _, remoteAddr := range order { + if aggregate := aggregates[remoteAddr]; aggregate != nil { + result = append(result, *aggregate) + } + } + return result, nil +} + +func mergeWAFIPStatusCodeCounts(ctx context.Context, filter NodeAccessLogFilter, aggregates map[string]*NodeAccessLogWAFIPAggregate) error { + if len(aggregates) == 0 { + return nil + } + conn, err := nodeAccessLogConn() + if err != nil { + return err + } + clause, args := buildNodeAccessLogFilterClause(filter) + tableName := nodeAccessLogTableName() + sql := fmt.Sprintf(` +SELECT + trim(remote_addr) AS trimmed_remote_addr, + status_code, + count() AS status_count +FROM %s +WHERE %s AND trim(remote_addr) != '' +GROUP BY trimmed_remote_addr, status_code`, tableName, clause) + rows, err := conn.Query(ctx, sql, args...) + if err != nil { + return fmt.Errorf("waf ip status code counts: %w", err) + } + defer func() { _ = rows.Close() }() + for rows.Next() { + var ( + remoteAddr string + statusCode int32 + statusCount uint64 + ) + if err := rows.Scan(&remoteAddr, &statusCode, &statusCount); err != nil { + return fmt.Errorf("scan waf ip status code row: %w", err) + } + remoteAddr = strings.TrimSpace(remoteAddr) + aggregate := aggregates[remoteAddr] + if aggregate == nil { + continue + } + if aggregate.StatusCounts == nil { + aggregate.StatusCounts = make(map[int]int64) + } + aggregate.StatusCounts[int(statusCode)] = safeInt64Count(statusCount) + } + return nil +} + // IPTrendNodeAccessLogs returns IP trend bucket rows. func IPTrendNodeAccessLogs(ctx context.Context, filter NodeAccessLogFilter, bucketSeconds int64) ([]NodeAccessLogIPTrend, error) { conn, err := nodeAccessLogConn()