固化公网流量汇总链路
This commit is contained in:
chenpei
parent
85c545eeb0
commit
2047e4eb2e
@ -82,6 +82,8 @@ RUN sed -i "s|https://dl-cdn.alpinelinux.org/alpine|${ALPINE_MIRROR}|g" /etc/apk
|
||||
RUN addgroup --system --gid 1001 nodejs && \
|
||||
adduser --system --uid 1001 nextjs
|
||||
|
||||
RUN mkdir -p /app/scripts
|
||||
|
||||
# 复制必要文件
|
||||
COPY --from=builder --chown=nextjs:nodejs /app/.next/standalone ./
|
||||
COPY --from=builder --chown=nextjs:nodejs /app/.next/static ./.next/static
|
||||
@ -98,9 +100,11 @@ COPY --from=builder --chown=nextjs:nodejs /app/node_modules/@smithy ./node_modul
|
||||
# 只复制 schema.prisma，不复制 migrations 和 engine-cache（减少 35MB）
|
||||
COPY --from=builder --chown=nextjs:nodejs /app/prisma/schema.prisma ./prisma/schema.prisma
|
||||
COPY --from=builder --chown=nextjs:nodejs /app/lib ./lib
|
||||
COPY --from=builder --chown=nextjs:nodejs /app/scripts/public-traffic-rollup-from-nginx.py ./scripts/public-traffic-rollup-from-nginx.py
|
||||
|
||||
# 创建必要目录并设置权限
|
||||
RUN mkdir -p /app/public /app/uploads && \
|
||||
mkdir -p /app/scripts && \
|
||||
mkdir -p /app/uploads/temp && \
|
||||
mkdir -p /app/uploads/assembled && \
|
||||
mkdir -p /app/uploads/materials && \
|
||||
|
||||
49
app/api/internal/public-traffic/rollups/route.ts
Normal file
49
app/api/internal/public-traffic/rollups/route.ts
Normal file
@ -0,0 +1,49 @@
|
||||
import { NextRequest, NextResponse } from 'next/server'
|
||||
import {
|
||||
upsertPublicTrafficRollups,
|
||||
type PublicTrafficRollupInput,
|
||||
} from '@/lib/public-traffic-stats'
|
||||
|
||||
export const runtime = 'nodejs'
|
||||
export const dynamic = 'force-dynamic'
|
||||
|
||||
const MAX_ROWS_PER_REQUEST = 2000
|
||||
|
||||
function readInternalToken(request: NextRequest) {
|
||||
const bearer = request.headers.get('authorization')?.match(/^Bearer\s+(.+)$/i)?.[1]
|
||||
return bearer || request.headers.get('x-internal-job-token') || ''
|
||||
}
|
||||
|
||||
function normalizeRows(payload: unknown): PublicTrafficRollupInput[] {
|
||||
if (!payload || typeof payload !== 'object') return []
|
||||
const rows = (payload as { rows?: unknown }).rows
|
||||
return Array.isArray(rows) ? rows.slice(0, MAX_ROWS_PER_REQUEST) as PublicTrafficRollupInput[] : []
|
||||
}
|
||||
|
||||
export async function POST(request: NextRequest) {
|
||||
const expectedToken = process.env.PUBLIC_TRAFFIC_ROLLUP_TOKEN
|
||||
|
||||
if (!expectedToken) {
|
||||
return NextResponse.json({ error: '公网流量汇总写入未启用' }, { status: 404 })
|
||||
}
|
||||
|
||||
if (readInternalToken(request) !== expectedToken) {
|
||||
return NextResponse.json({ error: '无权限写入公网流量汇总' }, { status: 403 })
|
||||
}
|
||||
|
||||
const payload = await request.json().catch(() => null)
|
||||
const rows = normalizeRows(payload)
|
||||
|
||||
if (rows.length === 0) {
|
||||
return NextResponse.json({ error: '没有可写入的公网流量汇总行' }, { status: 400 })
|
||||
}
|
||||
|
||||
const startedAt = Date.now()
|
||||
const result = await upsertPublicTrafficRollups(rows)
|
||||
|
||||
return NextResponse.json({
|
||||
ok: true,
|
||||
durationMs: Date.now() - startedAt,
|
||||
...result,
|
||||
})
|
||||
}
|
||||
@ -3,6 +3,7 @@ import os from 'os'
|
||||
import path from 'path'
|
||||
import {
|
||||
classifyPublicTrafficPurpose,
|
||||
getPublicTrafficPresetRange,
|
||||
getPublicTrafficStats,
|
||||
normalizeTrafficRange,
|
||||
parsePublicTrafficLogLine,
|
||||
@ -52,6 +53,20 @@ describe('public traffic stats', () => {
|
||||
expect(range.end.toISOString()).toBe('2026-07-04T15:59:59.999Z')
|
||||
})
|
||||
|
||||
test('builds public traffic presets using China business dates', () => {
|
||||
const now = new Date('2026-07-08T05:00:00.000Z')
|
||||
const today = getPublicTrafficPresetRange('today', now)
|
||||
const last7 = getPublicTrafficPresetRange('last7', now)
|
||||
const last30 = getPublicTrafficPresetRange('last30', now)
|
||||
|
||||
expect(today.start.toISOString()).toBe('2026-07-07T16:00:00.000Z')
|
||||
expect(today.end.toISOString()).toBe('2026-07-08T15:59:59.999Z')
|
||||
expect(last7.start.toISOString()).toBe('2026-07-01T16:00:00.000Z')
|
||||
expect(last7.end.toISOString()).toBe('2026-07-08T15:59:59.999Z')
|
||||
expect(last30.start.toISOString()).toBe('2026-06-08T16:00:00.000Z')
|
||||
expect(last30.end.toISOString()).toBe('2026-07-08T15:59:59.999Z')
|
||||
})
|
||||
|
||||
test('groups daily traffic by China business date', async () => {
|
||||
const previousPaths = process.env.PUBLIC_TRAFFIC_LOG_PATHS
|
||||
const tempDir = mkdtempSync(path.join(os.tmpdir(), 'public-traffic-'))
|
||||
|
||||
@ -123,6 +123,34 @@ interface PublicTrafficStatsOptions {
|
||||
forceRefresh?: boolean
|
||||
}
|
||||
|
||||
export type PublicTrafficPreset = 'today' | 'last7' | 'last30'
|
||||
|
||||
export type PublicTrafficRollupGranularity = '5m' | 'day'
|
||||
|
||||
export interface PublicTrafficRollupInput {
|
||||
bucketStart: string
|
||||
bucketEnd: string
|
||||
businessDate?: string | null
|
||||
granularity: PublicTrafficRollupGranularity
|
||||
domain: string
|
||||
method?: string | null
|
||||
path: string
|
||||
responseBytes: number
|
||||
requestBytes?: number | null
|
||||
requestBytesKnownCount?: number | null
|
||||
totalRequests: number
|
||||
statusCounts?: Partial<Record<TrafficStatusBucket, number>> | null
|
||||
sourceServer?: string | null
|
||||
generatedAt?: string | null
|
||||
frozen?: boolean | null
|
||||
}
|
||||
|
||||
export interface PublicTrafficRollupImportResult {
|
||||
received: number
|
||||
accepted: number
|
||||
skipped: number
|
||||
}
|
||||
|
||||
interface PublicTrafficScanContext {
|
||||
logPaths: string[]
|
||||
maxLines: number
|
||||
@ -137,6 +165,29 @@ interface PersistentTrafficCacheRow {
|
||||
expires_at: Date | string
|
||||
}
|
||||
|
||||
interface PublicTrafficRollupRow {
|
||||
bucket_start: Date | string
|
||||
bucket_end: Date | string
|
||||
business_date: string
|
||||
granularity: PublicTrafficRollupGranularity
|
||||
domain: string
|
||||
method: string
|
||||
path: string
|
||||
purpose_key: string
|
||||
purpose_label: string
|
||||
purpose_category: string
|
||||
purpose_detail: string
|
||||
response_bytes: bigint | number | string
|
||||
request_bytes: bigint | number | string | null
|
||||
request_bytes_known_count: number
|
||||
total_requests: number
|
||||
status_2xx: number
|
||||
status_3xx: number
|
||||
status_4xx: number
|
||||
status_5xx: number
|
||||
status_other: number
|
||||
}
|
||||
|
||||
const MONTHS: Record<string, number> = {
|
||||
Jan: 0,
|
||||
Feb: 1,
|
||||
@ -165,6 +216,7 @@ const statsCache = new Map<string, PublicTrafficCacheEntry>()
|
||||
const MAX_CACHE_ENTRIES = 20
|
||||
const refreshPromises = new Map<string, Promise<void>>()
|
||||
let persistentCacheTableEnsured = false
|
||||
let rollupTableEnsured = false
|
||||
|
||||
function emptyStatusCounts(): Record<TrafficStatusBucket, number> {
|
||||
return {
|
||||
@ -218,6 +270,26 @@ export function normalizeTrafficRange(startValue?: string | null, endValue?: str
|
||||
return { start, end }
|
||||
}
|
||||
|
||||
function toChinaDateString(date: Date): string {
|
||||
return new Intl.DateTimeFormat('en-CA', {
|
||||
timeZone: 'Asia/Shanghai',
|
||||
year: 'numeric',
|
||||
month: '2-digit',
|
||||
day: '2-digit',
|
||||
}).format(date)
|
||||
}
|
||||
|
||||
function addDays(date: Date, days: number): Date {
|
||||
return new Date(date.getTime() + days * 24 * 60 * 60 * 1000)
|
||||
}
|
||||
|
||||
export function getPublicTrafficPresetRange(preset: PublicTrafficPreset, now = new Date()): PublicTrafficRange {
|
||||
const endDate = toChinaDateString(now)
|
||||
const daysBack = preset === 'today' ? 0 : preset === 'last7' ? 6 : 29
|
||||
const startDate = toChinaDateString(addDays(now, -daysBack))
|
||||
return normalizeTrafficRange(startDate, endDate)
|
||||
}
|
||||
|
||||
function parseNginxTime(value: string): Date | null {
|
||||
const match = value.match(/^(\d{1,2})\/([A-Za-z]{3})\/(\d{4}):(\d{2}):(\d{2}):(\d{2}) ([+-]\d{4})$/)
|
||||
if (!match) return null
|
||||
@ -941,6 +1013,486 @@ async function writePersistentTrafficCache(
|
||||
)
|
||||
}
|
||||
|
||||
function bigintToNumber(value: bigint | number | string | null | undefined): number {
|
||||
if (value === null || value === undefined) return 0
|
||||
if (typeof value === 'bigint') return Number(value)
|
||||
if (typeof value === 'number') return Number.isFinite(value) ? value : 0
|
||||
const parsed = Number(value)
|
||||
return Number.isFinite(parsed) ? parsed : 0
|
||||
}
|
||||
|
||||
function normalizeStatusCounts(
|
||||
counts?: Partial<Record<TrafficStatusBucket, number>> | null,
|
||||
): Record<TrafficStatusBucket, number> {
|
||||
return {
|
||||
'2xx': Math.max(0, Math.floor(Number(counts?.['2xx'] || 0))),
|
||||
'3xx': Math.max(0, Math.floor(Number(counts?.['3xx'] || 0))),
|
||||
'4xx': Math.max(0, Math.floor(Number(counts?.['4xx'] || 0))),
|
||||
'5xx': Math.max(0, Math.floor(Number(counts?.['5xx'] || 0))),
|
||||
other: Math.max(0, Math.floor(Number(counts?.other || 0))),
|
||||
}
|
||||
}
|
||||
|
||||
function normalizeRollupInput(input: PublicTrafficRollupInput) {
|
||||
const bucketStart = new Date(input.bucketStart)
|
||||
const bucketEnd = new Date(input.bucketEnd)
|
||||
const domain = String(input.domain || '').trim().toLowerCase()
|
||||
const method = String(input.method || 'GET').trim().toUpperCase() || 'GET'
|
||||
const pathValue = normalizeRequestTarget(`${method} ${String(input.path || '').trim() || '/'} HTTP/1.1`).path
|
||||
const responseBytes = Math.max(0, Math.floor(Number(input.responseBytes || 0)))
|
||||
const totalRequests = Math.max(0, Math.floor(Number(input.totalRequests || 0)))
|
||||
const requestBytes =
|
||||
input.requestBytes === null || input.requestBytes === undefined
|
||||
? null
|
||||
: Math.max(0, Math.floor(Number(input.requestBytes || 0)))
|
||||
const requestBytesKnownCount = Math.max(0, Math.floor(Number(input.requestBytesKnownCount || 0)))
|
||||
const statusCounts = normalizeStatusCounts(input.statusCounts)
|
||||
|
||||
if (
|
||||
Number.isNaN(bucketStart.getTime()) ||
|
||||
Number.isNaN(bucketEnd.getTime()) ||
|
||||
bucketStart >= bucketEnd ||
|
||||
!domain ||
|
||||
!pathValue ||
|
||||
totalRequests <= 0 ||
|
||||
!['5m', 'day'].includes(input.granularity)
|
||||
) {
|
||||
return null
|
||||
}
|
||||
|
||||
const purpose = classifyPublicTrafficPurpose(domain, pathValue, method)
|
||||
const businessDate = input.businessDate && /^\d{4}-\d{2}-\d{2}$/.test(input.businessDate)
|
||||
? input.businessDate
|
||||
: toChinaDateString(bucketStart)
|
||||
const generatedAt = input.generatedAt ? new Date(input.generatedAt) : new Date()
|
||||
|
||||
return {
|
||||
bucketStart: bucketStart.toISOString(),
|
||||
bucketEnd: bucketEnd.toISOString(),
|
||||
businessDate,
|
||||
granularity: input.granularity,
|
||||
domain,
|
||||
method,
|
||||
path: pathValue,
|
||||
purposeKey: purpose.key,
|
||||
purposeLabel: purpose.label,
|
||||
purposeCategory: purpose.category,
|
||||
purposeDetail: purpose.detail,
|
||||
responseBytes,
|
||||
requestBytes,
|
||||
requestBytesKnownCount,
|
||||
totalRequests,
|
||||
status2xx: statusCounts['2xx'],
|
||||
status3xx: statusCounts['3xx'],
|
||||
status4xx: statusCounts['4xx'],
|
||||
status5xx: statusCounts['5xx'],
|
||||
statusOther: statusCounts.other,
|
||||
sourceServer: String(input.sourceServer || '').trim(),
|
||||
generatedAt: Number.isNaN(generatedAt.getTime()) ? new Date().toISOString() : generatedAt.toISOString(),
|
||||
frozen: Boolean(input.frozen),
|
||||
}
|
||||
}
|
||||
|
||||
async function ensurePublicTrafficRollupTable() {
|
||||
if (rollupTableEnsured) return true
|
||||
if (!persistentTrafficCacheEnabled()) return false
|
||||
|
||||
const { prisma } = await import('./prisma')
|
||||
|
||||
await prisma.$executeRawUnsafe(`
|
||||
CREATE TABLE IF NOT EXISTS "public_traffic_rollups" (
|
||||
"id" BIGSERIAL PRIMARY KEY,
|
||||
"bucket_start" TIMESTAMPTZ NOT NULL,
|
||||
"bucket_end" TIMESTAMPTZ NOT NULL,
|
||||
"business_date" TEXT NOT NULL,
|
||||
"granularity" TEXT NOT NULL,
|
||||
"domain" TEXT NOT NULL,
|
||||
"method" TEXT NOT NULL,
|
||||
"path" TEXT NOT NULL,
|
||||
"purpose_key" TEXT NOT NULL,
|
||||
"purpose_label" TEXT NOT NULL,
|
||||
"purpose_category" TEXT NOT NULL,
|
||||
"purpose_detail" TEXT NOT NULL,
|
||||
"response_bytes" BIGINT NOT NULL DEFAULT 0,
|
||||
"request_bytes" BIGINT,
|
||||
"request_bytes_known_count" INTEGER NOT NULL DEFAULT 0,
|
||||
"total_requests" INTEGER NOT NULL DEFAULT 0,
|
||||
"status_2xx" INTEGER NOT NULL DEFAULT 0,
|
||||
"status_3xx" INTEGER NOT NULL DEFAULT 0,
|
||||
"status_4xx" INTEGER NOT NULL DEFAULT 0,
|
||||
"status_5xx" INTEGER NOT NULL DEFAULT 0,
|
||||
"status_other" INTEGER NOT NULL DEFAULT 0,
|
||||
"source_server" TEXT NOT NULL DEFAULT '',
|
||||
"generated_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
"frozen" BOOLEAN NOT NULL DEFAULT false,
|
||||
"created_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
"updated_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
CONSTRAINT "public_traffic_rollups_unique_bucket_path"
|
||||
UNIQUE ("business_date", "granularity", "bucket_start", "bucket_end", "domain", "method", "path")
|
||||
)
|
||||
`)
|
||||
await prisma.$executeRawUnsafe(`
|
||||
CREATE INDEX IF NOT EXISTS "public_traffic_rollups_date_granularity_idx"
|
||||
ON "public_traffic_rollups" ("business_date", "granularity")
|
||||
`)
|
||||
await prisma.$executeRawUnsafe(`
|
||||
CREATE INDEX IF NOT EXISTS "public_traffic_rollups_bucket_idx"
|
||||
ON "public_traffic_rollups" ("bucket_start", "bucket_end")
|
||||
`)
|
||||
|
||||
rollupTableEnsured = true
|
||||
return true
|
||||
}
|
||||
|
||||
export async function upsertPublicTrafficRollups(
|
||||
inputs: PublicTrafficRollupInput[],
|
||||
): Promise<PublicTrafficRollupImportResult> {
|
||||
if (!(await ensurePublicTrafficRollupTable())) {
|
||||
throw new Error('公网流量汇总表未启用')
|
||||
}
|
||||
|
||||
const rows: Array<NonNullable<ReturnType<typeof normalizeRollupInput>>> = []
|
||||
for (const input of inputs) {
|
||||
const row = normalizeRollupInput(input)
|
||||
if (row) rows.push(row)
|
||||
}
|
||||
|
||||
if (rows.length === 0) {
|
||||
return { received: inputs.length, accepted: 0, skipped: inputs.length }
|
||||
}
|
||||
|
||||
const { prisma } = await import('./prisma')
|
||||
const chunkSize = 500
|
||||
|
||||
for (let index = 0; index < rows.length; index += chunkSize) {
|
||||
const chunk = rows.slice(index, index + chunkSize)
|
||||
await prisma.$executeRawUnsafe(
|
||||
`
|
||||
INSERT INTO "public_traffic_rollups" (
|
||||
"bucket_start",
|
||||
"bucket_end",
|
||||
"business_date",
|
||||
"granularity",
|
||||
"domain",
|
||||
"method",
|
||||
"path",
|
||||
"purpose_key",
|
||||
"purpose_label",
|
||||
"purpose_category",
|
||||
"purpose_detail",
|
||||
"response_bytes",
|
||||
"request_bytes",
|
||||
"request_bytes_known_count",
|
||||
"total_requests",
|
||||
"status_2xx",
|
||||
"status_3xx",
|
||||
"status_4xx",
|
||||
"status_5xx",
|
||||
"status_other",
|
||||
"source_server",
|
||||
"generated_at",
|
||||
"frozen",
|
||||
"updated_at"
|
||||
)
|
||||
SELECT
|
||||
"bucketStart"::timestamptz,
|
||||
"bucketEnd"::timestamptz,
|
||||
"businessDate",
|
||||
"granularity",
|
||||
"domain",
|
||||
"method",
|
||||
"path",
|
||||
"purposeKey",
|
||||
"purposeLabel",
|
||||
"purposeCategory",
|
||||
"purposeDetail",
|
||||
"responseBytes"::bigint,
|
||||
NULLIF("requestBytes", '')::bigint,
|
||||
"requestBytesKnownCount"::integer,
|
||||
"totalRequests"::integer,
|
||||
"status2xx"::integer,
|
||||
"status3xx"::integer,
|
||||
"status4xx"::integer,
|
||||
"status5xx"::integer,
|
||||
"statusOther"::integer,
|
||||
"sourceServer",
|
||||
"generatedAt"::timestamptz,
|
||||
"frozen"::boolean,
|
||||
CURRENT_TIMESTAMP
|
||||
FROM jsonb_to_recordset($1::jsonb) AS x(
|
||||
"bucketStart" text,
|
||||
"bucketEnd" text,
|
||||
"businessDate" text,
|
||||
"granularity" text,
|
||||
"domain" text,
|
||||
"method" text,
|
||||
"path" text,
|
||||
"purposeKey" text,
|
||||
"purposeLabel" text,
|
||||
"purposeCategory" text,
|
||||
"purposeDetail" text,
|
||||
"responseBytes" text,
|
||||
"requestBytes" text,
|
||||
"requestBytesKnownCount" text,
|
||||
"totalRequests" text,
|
||||
"status2xx" text,
|
||||
"status3xx" text,
|
||||
"status4xx" text,
|
||||
"status5xx" text,
|
||||
"statusOther" text,
|
||||
"sourceServer" text,
|
||||
"generatedAt" text,
|
||||
"frozen" text
|
||||
)
|
||||
ON CONFLICT ("business_date", "granularity", "bucket_start", "bucket_end", "domain", "method", "path")
|
||||
DO UPDATE SET
|
||||
"purpose_key" = EXCLUDED."purpose_key",
|
||||
"purpose_label" = EXCLUDED."purpose_label",
|
||||
"purpose_category" = EXCLUDED."purpose_category",
|
||||
"purpose_detail" = EXCLUDED."purpose_detail",
|
||||
"response_bytes" = EXCLUDED."response_bytes",
|
||||
"request_bytes" = EXCLUDED."request_bytes",
|
||||
"request_bytes_known_count" = EXCLUDED."request_bytes_known_count",
|
||||
"total_requests" = EXCLUDED."total_requests",
|
||||
"status_2xx" = EXCLUDED."status_2xx",
|
||||
"status_3xx" = EXCLUDED."status_3xx",
|
||||
"status_4xx" = EXCLUDED."status_4xx",
|
||||
"status_5xx" = EXCLUDED."status_5xx",
|
||||
"status_other" = EXCLUDED."status_other",
|
||||
"source_server" = EXCLUDED."source_server",
|
||||
"generated_at" = EXCLUDED."generated_at",
|
||||
"frozen" = EXCLUDED."frozen",
|
||||
"updated_at" = CURRENT_TIMESTAMP
|
||||
`,
|
||||
JSON.stringify(
|
||||
chunk.map((row) => ({
|
||||
...row,
|
||||
requestBytes: row.requestBytes === null ? '' : String(row.requestBytes),
|
||||
responseBytes: String(row.responseBytes),
|
||||
requestBytesKnownCount: String(row.requestBytesKnownCount),
|
||||
totalRequests: String(row.totalRequests),
|
||||
status2xx: String(row.status2xx),
|
||||
status3xx: String(row.status3xx),
|
||||
status4xx: String(row.status4xx),
|
||||
status5xx: String(row.status5xx),
|
||||
statusOther: String(row.statusOther),
|
||||
frozen: String(row.frozen),
|
||||
})),
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
return { received: inputs.length, accepted: rows.length, skipped: inputs.length - rows.length }
|
||||
}
|
||||
|
||||
function addRollupRow(stats: PublicTrafficStats, row: PublicTrafficRollupRow) {
|
||||
const responseBytes = bigintToNumber(row.response_bytes)
|
||||
const requestBytes = row.request_bytes === null ? null : bigintToNumber(row.request_bytes)
|
||||
const requestBytesKnownCount = Number(row.request_bytes_known_count || 0)
|
||||
const totalRequests = Number(row.total_requests || 0)
|
||||
const bucketStart = row.bucket_start instanceof Date ? row.bucket_start : new Date(row.bucket_start)
|
||||
const bucketEnd = row.bucket_end instanceof Date ? row.bucket_end : new Date(row.bucket_end)
|
||||
const statusCounts: Record<TrafficStatusBucket, number> = {
|
||||
'2xx': Number(row.status_2xx || 0),
|
||||
'3xx': Number(row.status_3xx || 0),
|
||||
'4xx': Number(row.status_4xx || 0),
|
||||
'5xx': Number(row.status_5xx || 0),
|
||||
other: Number(row.status_other || 0),
|
||||
}
|
||||
|
||||
stats.summary.responseBytes += responseBytes
|
||||
if (requestBytesKnownCount > 0) {
|
||||
stats.summary.requestBytes = (stats.summary.requestBytes || 0) + (requestBytes || 0)
|
||||
stats.summary.requestBytesKnownCount += requestBytesKnownCount
|
||||
}
|
||||
stats.summary.totalRequests += totalRequests
|
||||
for (const bucket of Object.keys(statusCounts) as TrafficStatusBucket[]) {
|
||||
stats.summary.statusCounts[bucket] += statusCounts[bucket]
|
||||
}
|
||||
|
||||
const firstSeen = Number.isNaN(bucketStart.getTime()) ? null : bucketStart.toISOString()
|
||||
const lastSeen = Number.isNaN(bucketEnd.getTime()) ? null : bucketEnd.toISOString()
|
||||
if (firstSeen && (!stats.summary.firstSeen || firstSeen < stats.summary.firstSeen)) stats.summary.firstSeen = firstSeen
|
||||
if (lastSeen && (!stats.summary.lastSeen || lastSeen > stats.summary.lastSeen)) stats.summary.lastSeen = lastSeen
|
||||
|
||||
let dayItem = stats.byDay.find((item) => item.date === row.business_date)
|
||||
if (!dayItem) {
|
||||
dayItem = {
|
||||
date: row.business_date,
|
||||
responseBytes: 0,
|
||||
requestBytes: null,
|
||||
requestBytesKnownCount: 0,
|
||||
totalRequests: 0,
|
||||
}
|
||||
stats.byDay.push(dayItem)
|
||||
}
|
||||
dayItem.responseBytes += responseBytes
|
||||
dayItem.totalRequests += totalRequests
|
||||
if (requestBytesKnownCount > 0) {
|
||||
dayItem.requestBytes = (dayItem.requestBytes || 0) + (requestBytes || 0)
|
||||
dayItem.requestBytesKnownCount += requestBytesKnownCount
|
||||
}
|
||||
|
||||
let domainItem = stats.byDomain.find((item) => item.domain === row.domain)
|
||||
if (!domainItem) {
|
||||
domainItem = {
|
||||
domain: row.domain,
|
||||
responseBytes: 0,
|
||||
requestBytes: null,
|
||||
requestBytesKnownCount: 0,
|
||||
totalRequests: 0,
|
||||
statusCounts: emptyStatusCounts(),
|
||||
}
|
||||
stats.byDomain.push(domainItem)
|
||||
}
|
||||
domainItem.responseBytes += responseBytes
|
||||
domainItem.totalRequests += totalRequests
|
||||
if (requestBytesKnownCount > 0) {
|
||||
domainItem.requestBytes = (domainItem.requestBytes || 0) + (requestBytes || 0)
|
||||
domainItem.requestBytesKnownCount += requestBytesKnownCount
|
||||
}
|
||||
for (const bucket of Object.keys(statusCounts) as TrafficStatusBucket[]) {
|
||||
domainItem.statusCounts[bucket] += statusCounts[bucket]
|
||||
}
|
||||
|
||||
let purposeItem = stats.byPurpose.find((item) => item.key === row.purpose_key)
|
||||
if (!purposeItem) {
|
||||
purposeItem = {
|
||||
key: row.purpose_key,
|
||||
label: row.purpose_label,
|
||||
category: row.purpose_category,
|
||||
detail: row.purpose_detail,
|
||||
responseBytes: 0,
|
||||
requestBytes: null,
|
||||
requestBytesKnownCount: 0,
|
||||
totalRequests: 0,
|
||||
}
|
||||
stats.byPurpose.push(purposeItem)
|
||||
}
|
||||
purposeItem.responseBytes += responseBytes
|
||||
purposeItem.totalRequests += totalRequests
|
||||
if (requestBytesKnownCount > 0) {
|
||||
purposeItem.requestBytes = (purposeItem.requestBytes || 0) + (requestBytes || 0)
|
||||
purposeItem.requestBytesKnownCount += requestBytesKnownCount
|
||||
}
|
||||
|
||||
let topPathItem = stats.topPaths.find((item) => item.method === row.method && item.path === row.path)
|
||||
if (!topPathItem) {
|
||||
topPathItem = {
|
||||
path: row.path,
|
||||
method: row.method,
|
||||
purposeLabel: row.purpose_label,
|
||||
purposeCategory: row.purpose_category,
|
||||
purposeDetail: row.purpose_detail,
|
||||
responseBytes: 0,
|
||||
totalRequests: 0,
|
||||
}
|
||||
stats.topPaths.push(topPathItem)
|
||||
}
|
||||
topPathItem.responseBytes += responseBytes
|
||||
topPathItem.totalRequests += totalRequests
|
||||
}
|
||||
|
||||
async function getPublicTrafficStatsFromRollups(range: PublicTrafficRange): Promise<PublicTrafficStats | null> {
|
||||
if (!(await ensurePublicTrafficRollupTable())) return null
|
||||
|
||||
const startDate = toChinaDateString(range.start)
|
||||
const endDate = toChinaDateString(range.end)
|
||||
const todayDate = toChinaDateString(new Date())
|
||||
const { prisma } = await import('./prisma')
|
||||
const rows = await prisma.$queryRawUnsafe<PublicTrafficRollupRow[]>(
|
||||
`
|
||||
SELECT
|
||||
"bucket_start",
|
||||
"bucket_end",
|
||||
"business_date",
|
||||
"granularity",
|
||||
"domain",
|
||||
"method",
|
||||
"path",
|
||||
"purpose_key",
|
||||
"purpose_label",
|
||||
"purpose_category",
|
||||
"purpose_detail",
|
||||
"response_bytes",
|
||||
"request_bytes",
|
||||
"request_bytes_known_count",
|
||||
"total_requests",
|
||||
"status_2xx",
|
||||
"status_3xx",
|
||||
"status_4xx",
|
||||
"status_5xx",
|
||||
"status_other"
|
||||
FROM "public_traffic_rollups"
|
||||
WHERE "business_date" >= $1
|
||||
AND "business_date" <= $2
|
||||
AND (
|
||||
("business_date" = $3 AND "granularity" = '5m')
|
||||
OR ("business_date" <> $3 AND "granularity" = 'day')
|
||||
)
|
||||
ORDER BY "bucket_start" ASC
|
||||
`,
|
||||
startDate,
|
||||
endDate,
|
||||
todayDate,
|
||||
)
|
||||
|
||||
if (!rows.length) return null
|
||||
|
||||
const stats: PublicTrafficStats = {
|
||||
generatedAt: new Date().toISOString(),
|
||||
available: true,
|
||||
range: {
|
||||
start: range.start.toISOString(),
|
||||
end: range.end.toISOString(),
|
||||
},
|
||||
summary: {
|
||||
responseBytes: 0,
|
||||
requestBytes: null,
|
||||
requestBytesKnownCount: 0,
|
||||
totalRequests: 0,
|
||||
uniqueIpCount: 0,
|
||||
statusCounts: emptyStatusCounts(),
|
||||
firstSeen: null,
|
||||
lastSeen: null,
|
||||
},
|
||||
byDay: [],
|
||||
byDomain: [],
|
||||
byPurpose: [],
|
||||
topPaths: [],
|
||||
source: {
|
||||
logPaths: [],
|
||||
scannedFiles: 0,
|
||||
scannedLines: 0,
|
||||
matchedLines: 0,
|
||||
skippedLines: 0,
|
||||
truncated: false,
|
||||
issues: [],
|
||||
requestBytesAvailable: false,
|
||||
cache: {
|
||||
persisted: true,
|
||||
hit: true,
|
||||
stale: false,
|
||||
refreshInProgress: false,
|
||||
refreshedAt: null,
|
||||
expiresAt: null,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for (const row of rows) addRollupRow(stats, row)
|
||||
|
||||
stats.source.matchedLines = stats.summary.totalRequests
|
||||
stats.source.requestBytesAvailable = stats.summary.requestBytesKnownCount > 0
|
||||
stats.byDay.sort((a, b) => a.date.localeCompare(b.date))
|
||||
stats.byDomain.sort((a, b) => b.responseBytes - a.responseBytes)
|
||||
stats.byPurpose.sort((a, b) => b.responseBytes - a.responseBytes)
|
||||
stats.topPaths.sort((a, b) => b.responseBytes - a.responseBytes)
|
||||
stats.topPaths = stats.topPaths.slice(0, 20)
|
||||
|
||||
return stats
|
||||
}
|
||||
|
||||
function getCachedStats(cacheKey: string): PublicTrafficStats | null {
|
||||
const cached = statsCache.get(cacheKey)
|
||||
if (!cached) return null
|
||||
@ -1061,6 +1613,13 @@ export async function getPublicTrafficStats(
|
||||
range: PublicTrafficRange,
|
||||
options: PublicTrafficStatsOptions = {},
|
||||
): Promise<PublicTrafficStats> {
|
||||
try {
|
||||
const rollupStats = await getPublicTrafficStatsFromRollups(range)
|
||||
if (rollupStats) return rollupStats
|
||||
} catch (error) {
|
||||
console.error('[public-traffic] 读取公网流量汇总表失败:', error)
|
||||
}
|
||||
|
||||
const context = getScanContext(range)
|
||||
const cacheKey = context.stableCacheKey
|
||||
|
||||
|
||||
@ -528,6 +528,10 @@ start_container() {
|
||||
run_args+=(-e "DOWNLOAD_TOKEN_SECRET=$DOWNLOAD_TOKEN_SECRET")
|
||||
fi
|
||||
|
||||
if [ -n "${PUBLIC_TRAFFIC_ROLLUP_TOKEN:-}" ]; then
|
||||
run_args+=(-e "PUBLIC_TRAFFIC_ROLLUP_TOKEN=$PUBLIC_TRAFFIC_ROLLUP_TOKEN")
|
||||
fi
|
||||
|
||||
"${run_args[@]}" "$image"
|
||||
}
|
||||
|
||||
@ -557,6 +561,139 @@ PROXY_SERVICE="${CONTAINER_NAME}-local-proxy.service"
|
||||
CURRENT_FILE="${PROXY_ROOT}/current-container"
|
||||
BLUE_CONTAINER="${CONTAINER_NAME}-blue"
|
||||
GREEN_CONTAINER="${CONTAINER_NAME}-green"
|
||||
PUBLIC_TRAFFIC_SYNC_ROOT="/mnt/PoolOne/apps/youyidian-3d/public-traffic-sync"
|
||||
PUBLIC_TRAFFIC_ROLLUP_TOKEN_FILE="${PUBLIC_TRAFFIC_SYNC_ROOT}/rollup.token"
|
||||
PUBLIC_TRAFFIC_LOG_SYNC_SCRIPT="${PUBLIC_TRAFFIC_SYNC_ROOT}/public-traffic-log-sync.sh"
|
||||
PUBLIC_TRAFFIC_ROLLUP_SCRIPT="${PUBLIC_TRAFFIC_SYNC_ROOT}/public-traffic-rollup-sync.sh"
|
||||
PUBLIC_TRAFFIC_ROLLUP_SERVICE="public-traffic-rollup-sync.service"
|
||||
PUBLIC_TRAFFIC_ROLLUP_TIMER="public-traffic-rollup-sync.timer"
|
||||
PUBLIC_TRAFFIC_ROLLUP_TOKEN=""
|
||||
|
||||
ensure_public_traffic_rollup_token() {
|
||||
if [ "$TARGET" != "production" ]; then
|
||||
return 0
|
||||
fi
|
||||
|
||||
mkdir -p "$PUBLIC_TRAFFIC_SYNC_ROOT"
|
||||
chmod 700 "$PUBLIC_TRAFFIC_SYNC_ROOT"
|
||||
|
||||
if [ ! -s "$PUBLIC_TRAFFIC_ROLLUP_TOKEN_FILE" ]; then
|
||||
if command -v openssl >/dev/null 2>&1; then
|
||||
openssl rand -hex 32 > "$PUBLIC_TRAFFIC_ROLLUP_TOKEN_FILE"
|
||||
else
|
||||
tr -dc 'A-Za-z0-9' </dev/urandom | head -c 64 > "$PUBLIC_TRAFFIC_ROLLUP_TOKEN_FILE"
|
||||
fi
|
||||
fi
|
||||
|
||||
chmod 600 "$PUBLIC_TRAFFIC_ROLLUP_TOKEN_FILE"
|
||||
PUBLIC_TRAFFIC_ROLLUP_TOKEN="$(tr -d '\r\n' < "$PUBLIC_TRAFFIC_ROLLUP_TOKEN_FILE")"
|
||||
}
|
||||
|
||||
install_public_traffic_rollup_timers() {
|
||||
if [ "$TARGET" != "production" ]; then
|
||||
return 0
|
||||
fi
|
||||
|
||||
mkdir -p "$PUBLIC_TRAFFIC_SYNC_ROOT"
|
||||
cat > "$PUBLIC_TRAFFIC_LOG_SYNC_SCRIPT" <<'EOF'
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
remote='trafficlog@218.93.9.38'
|
||||
key='/root/.ssh/public_traffic_log_sync_ed25519'
|
||||
dest='/mnt/PoolOne/apps/youyidian-3d/public-traffic-logs'
|
||||
ssh_opts="ssh -i ${key} -o BatchMode=yes -o StrictHostKeyChecking=yes -o ConnectTimeout=10"
|
||||
domains=(
|
||||
sc.jiansun.vip
|
||||
s3.jiansun.vip
|
||||
cp.jiansun.vip
|
||||
)
|
||||
mkdir -p "$dest"
|
||||
chmod 755 "$dest"
|
||||
status=0
|
||||
for domain in "${domains[@]}"; do
|
||||
if ! rsync -rtz --timeout=45 --chmod=F644,D755 -e "$ssh_opts" "${remote}:/var/log/nginx/${domain}.access.log*" "$dest/"; then
|
||||
echo "WARN: failed to sync ${domain}.access.log*" >&2
|
||||
status=1
|
||||
fi
|
||||
done
|
||||
find "$dest" -maxdepth 1 -type f -name '*.access.log*' -exec chmod 0644 {} +
|
||||
exit "$status"
|
||||
EOF
|
||||
chmod 700 "$PUBLIC_TRAFFIC_LOG_SYNC_SCRIPT"
|
||||
|
||||
cat > "/etc/systemd/system/public-traffic-log-sync.service" <<EOF
|
||||
[Unit]
|
||||
Description=Sync public traffic nginx logs from tunnel server
|
||||
After=network-online.target
|
||||
Wants=network-online.target
|
||||
|
||||
[Service]
|
||||
Type=oneshot
|
||||
ExecStart=${PUBLIC_TRAFFIC_LOG_SYNC_SCRIPT}
|
||||
TimeoutStartSec=180
|
||||
EOF
|
||||
|
||||
cat > "/etc/systemd/system/public-traffic-log-sync.timer" <<EOF
|
||||
[Unit]
|
||||
Description=Run public traffic log sync every 5 minutes
|
||||
|
||||
[Timer]
|
||||
OnBootSec=1min
|
||||
OnUnitActiveSec=5min
|
||||
Persistent=true
|
||||
|
||||
[Install]
|
||||
WantedBy=timers.target
|
||||
EOF
|
||||
|
||||
cat > "$PUBLIC_TRAFFIC_ROLLUP_SCRIPT" <<EOF
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
current_file='${CURRENT_FILE}'
|
||||
container="\$(cat "\$current_file" 2>/dev/null || true)"
|
||||
if [ -z "\$container" ] || ! docker inspect "\$container" >/dev/null 2>&1; then
|
||||
echo "No active material container found" >&2
|
||||
exit 1
|
||||
fi
|
||||
systemctl start public-traffic-log-sync.service >/dev/null 2>&1 || true
|
||||
docker exec "\$container" python3 /app/scripts/public-traffic-rollup-from-nginx.py --granularity 5m
|
||||
yesterday="\$(TZ=Asia/Shanghai date -d 'yesterday' +%F)"
|
||||
docker exec "\$container" python3 /app/scripts/public-traffic-rollup-from-nginx.py --granularity day --start-date "\$yesterday" --end-date "\$yesterday"
|
||||
EOF
|
||||
chmod 700 "$PUBLIC_TRAFFIC_ROLLUP_SCRIPT"
|
||||
|
||||
cat > "/etc/systemd/system/${PUBLIC_TRAFFIC_ROLLUP_SERVICE}" <<EOF
|
||||
[Unit]
|
||||
Description=Sync public traffic statistics rollups
|
||||
After=network-online.target ${PROXY_SERVICE}
|
||||
Wants=network-online.target
|
||||
|
||||
[Service]
|
||||
Type=oneshot
|
||||
ExecStart=${PUBLIC_TRAFFIC_ROLLUP_SCRIPT}
|
||||
TimeoutStartSec=360
|
||||
EOF
|
||||
|
||||
cat > "/etc/systemd/system/${PUBLIC_TRAFFIC_ROLLUP_TIMER}" <<EOF
|
||||
[Unit]
|
||||
Description=Run public traffic statistics rollup sync
|
||||
|
||||
[Timer]
|
||||
OnBootSec=2min
|
||||
OnUnitActiveSec=5min
|
||||
RandomizedDelaySec=30s
|
||||
Persistent=true
|
||||
|
||||
[Install]
|
||||
WantedBy=timers.target
|
||||
EOF
|
||||
|
||||
systemctl daemon-reload
|
||||
systemctl enable public-traffic-log-sync.timer >/dev/null 2>&1 || true
|
||||
systemctl enable "$PUBLIC_TRAFFIC_ROLLUP_TIMER" >/dev/null 2>&1 || true
|
||||
systemctl restart public-traffic-log-sync.timer >/dev/null 2>&1 || true
|
||||
systemctl restart "$PUBLIC_TRAFFIC_ROLLUP_TIMER" >/dev/null 2>&1 || true
|
||||
}
|
||||
|
||||
is_proxy_running() {
|
||||
systemctl is-active --quiet "$PROXY_SERVICE" 2>/dev/null || {
|
||||
@ -703,6 +840,8 @@ if is_proxy_running; then
|
||||
PROXY_WAS_RUNNING=1
|
||||
fi
|
||||
|
||||
ensure_public_traffic_rollup_token
|
||||
|
||||
log "Starting candidate container $NEXT_CONTAINER"
|
||||
docker rm -f "$NEXT_CONTAINER" >/dev/null 2>&1 || true
|
||||
start_container "$NEXT_CONTAINER" "$REGISTRY_COMMIT" 0 >/tmp/git-deploy-new-container-id.txt
|
||||
@ -762,6 +901,28 @@ printf '%s\n' "$NEXT_CONTAINER" > "$CURRENT_FILE"
|
||||
ok "Stable local entry is healthy on port $PORT"
|
||||
docker logs --tail 20 "$NEXT_CONTAINER" 2>&1 || true
|
||||
|
||||
if install_public_traffic_rollup_timers; then
|
||||
if [ "$TARGET" = "production" ]; then
|
||||
if systemctl start "$PUBLIC_TRAFFIC_ROLLUP_SERVICE" >/tmp/public-traffic-rollup-sync.log 2>/tmp/public-traffic-rollup-sync.err; then
|
||||
ok "Public traffic statistics rollups synced"
|
||||
else
|
||||
warn "Public traffic statistics rollup sync failed; page remains available, check /tmp/public-traffic-rollup-sync.err"
|
||||
cat /tmp/public-traffic-rollup-sync.err >&2 || true
|
||||
fi
|
||||
|
||||
backfill_start="$(TZ=Asia/Shanghai date -d '7 days ago' +%F)"
|
||||
backfill_end="$(TZ=Asia/Shanghai date -d 'yesterday' +%F)"
|
||||
if docker exec "$NEXT_CONTAINER" python3 /app/scripts/public-traffic-rollup-from-nginx.py --granularity day --start-date "$backfill_start" --end-date "$backfill_end" >/tmp/public-traffic-rollup-backfill.log 2>/tmp/public-traffic-rollup-backfill.err; then
|
||||
ok "Public traffic historical day rollups backfilled from ${backfill_start} to ${backfill_end}"
|
||||
else
|
||||
warn "Public traffic historical backfill failed; check /tmp/public-traffic-rollup-backfill.err"
|
||||
cat /tmp/public-traffic-rollup-backfill.err >&2 || true
|
||||
fi
|
||||
fi
|
||||
else
|
||||
warn "Public traffic statistics rollup timers were not installed"
|
||||
fi
|
||||
|
||||
if [ -n "$CURRENT_CONTAINER" ] && [ "$CURRENT_CONTAINER" != "$NEXT_CONTAINER" ]; then
|
||||
docker rm -f "$CURRENT_CONTAINER" >/dev/null 2>&1 || true
|
||||
fi
|
||||
|
||||
272
scripts/public-traffic-rollup-from-nginx.py
Normal file
272
scripts/public-traffic-rollup-from-nginx.py
Normal file
@ -0,0 +1,272 @@
|
||||
#!/usr/bin/env python3
|
||||
import argparse
|
||||
import gzip
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import socket
|
||||
import sys
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from collections import defaultdict
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
CHINA_TZ = timezone(timedelta(hours=8))
|
||||
DOMAINS = ("sc.jiansun.vip", "s3.jiansun.vip", "cp.jiansun.vip")
|
||||
LOG_RE = re.compile(
|
||||
r'^(?P<remote>\S+) \S+ \S+ \[(?P<time>[^\]]+)\] '
|
||||
r'"(?P<method>[A-Z]+) (?P<target>\S+) HTTP/[^"]+" '
|
||||
r'(?P<status>\d{3}) (?P<body_bytes>\S+)(?P<rest>.*)$'
|
||||
)
|
||||
|
||||
|
||||
def parse_args():
|
||||
parser = argparse.ArgumentParser(description="Aggregate public traffic nginx logs and push rollups.")
|
||||
parser.add_argument("--log-dir", default="/var/log/nginx")
|
||||
parser.add_argument("--endpoint", default="http://127.0.0.1:3000/api/internal/public-traffic/rollups")
|
||||
parser.add_argument("--token-env", default="PUBLIC_TRAFFIC_ROLLUP_TOKEN")
|
||||
parser.add_argument("--granularity", choices=("5m", "day"), required=True)
|
||||
parser.add_argument("--start-date", help="China business date, YYYY-MM-DD")
|
||||
parser.add_argument("--end-date", help="China business date, YYYY-MM-DD")
|
||||
parser.add_argument("--source-server", default=socket.gethostname())
|
||||
parser.add_argument("--batch-size", type=int, default=1000)
|
||||
parser.add_argument("--domains", default=",".join(DOMAINS))
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def parse_date(value):
|
||||
return datetime.strptime(value, "%Y-%m-%d").date()
|
||||
|
||||
|
||||
def default_dates(granularity):
|
||||
today = datetime.now(CHINA_TZ).date()
|
||||
if granularity == "5m":
|
||||
return today, today
|
||||
yesterday = today - timedelta(days=1)
|
||||
return yesterday, yesterday
|
||||
|
||||
|
||||
def iter_log_paths(log_dir, domains):
|
||||
root = Path(log_dir)
|
||||
paths = []
|
||||
for domain in domains:
|
||||
paths.extend(root.glob(f"{domain}.access.log*"))
|
||||
return sorted(paths, key=lambda item: item.name)
|
||||
|
||||
|
||||
def open_log(path):
|
||||
if str(path).endswith(".gz"):
|
||||
return gzip.open(path, "rt", encoding="utf-8", errors="replace")
|
||||
return open(path, "rt", encoding="utf-8", errors="replace")
|
||||
|
||||
|
||||
def infer_domain(path):
|
||||
name = Path(path).name
|
||||
for domain in DOMAINS:
|
||||
if name.startswith(f"{domain}."):
|
||||
return domain
|
||||
return name.split(".access.log", 1)[0]
|
||||
|
||||
|
||||
def normalize_target(target):
|
||||
try:
|
||||
parsed = urllib.parse.urlsplit(target)
|
||||
path = parsed.path or "/"
|
||||
except Exception:
|
||||
path = target.split("?", 1)[0] or "/"
|
||||
if not path.startswith("/"):
|
||||
path = "/" + path
|
||||
return path
|
||||
|
||||
|
||||
def parse_size(value):
|
||||
if value == "-":
|
||||
return 0
|
||||
try:
|
||||
return max(0, int(value))
|
||||
except ValueError:
|
||||
return 0
|
||||
|
||||
|
||||
def parse_extra_int(rest, key):
|
||||
match = re.search(rf"{re.escape(key)}=(\d+)", rest)
|
||||
return int(match.group(1)) if match else None
|
||||
|
||||
|
||||
def status_bucket(status):
|
||||
if 200 <= status <= 299:
|
||||
return "2xx"
|
||||
if 300 <= status <= 399:
|
||||
return "3xx"
|
||||
if 400 <= status <= 499:
|
||||
return "4xx"
|
||||
if 500 <= status <= 599:
|
||||
return "5xx"
|
||||
return "other"
|
||||
|
||||
|
||||
def parse_line(line, domain):
|
||||
match = LOG_RE.match(line)
|
||||
if not match:
|
||||
return None
|
||||
try:
|
||||
timestamp = datetime.strptime(match.group("time"), "%d/%b/%Y:%H:%M:%S %z").astimezone(CHINA_TZ)
|
||||
status = int(match.group("status"))
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
rest = match.group("rest") or ""
|
||||
response_bytes = parse_extra_int(rest, "bytes_sent")
|
||||
if response_bytes is None:
|
||||
response_bytes = parse_size(match.group("body_bytes"))
|
||||
request_bytes = parse_extra_int(rest, "request_length")
|
||||
|
||||
return {
|
||||
"timestamp": timestamp,
|
||||
"businessDate": timestamp.date().isoformat(),
|
||||
"domain": domain,
|
||||
"method": match.group("method").upper(),
|
||||
"path": normalize_target(match.group("target")),
|
||||
"statusBucket": status_bucket(status),
|
||||
"responseBytes": response_bytes,
|
||||
"requestBytes": request_bytes,
|
||||
}
|
||||
|
||||
|
||||
def bucket_for(timestamp, granularity):
|
||||
if granularity == "day":
|
||||
start = timestamp.replace(hour=0, minute=0, second=0, microsecond=0)
|
||||
end = start + timedelta(days=1)
|
||||
return start, end
|
||||
minute = (timestamp.minute // 5) * 5
|
||||
start = timestamp.replace(minute=minute, second=0, microsecond=0)
|
||||
end = start + timedelta(minutes=5)
|
||||
return start, end
|
||||
|
||||
|
||||
def iso_utc(value):
|
||||
return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
|
||||
|
||||
|
||||
def aggregate(args):
|
||||
domains = tuple(item.strip() for item in args.domains.split(",") if item.strip())
|
||||
start_default, end_default = default_dates(args.granularity)
|
||||
start_date = parse_date(args.start_date) if args.start_date else start_default
|
||||
end_date = parse_date(args.end_date) if args.end_date else end_default
|
||||
|
||||
rows = {}
|
||||
scanned_lines = 0
|
||||
matched_lines = 0
|
||||
for path in iter_log_paths(args.log_dir, domains):
|
||||
domain = infer_domain(path)
|
||||
try:
|
||||
handle = open_log(path)
|
||||
except OSError as exc:
|
||||
print(f"WARN: cannot open {path}: {exc}", file=sys.stderr)
|
||||
continue
|
||||
with handle:
|
||||
for line in handle:
|
||||
scanned_lines += 1
|
||||
entry = parse_line(line, domain)
|
||||
if not entry:
|
||||
continue
|
||||
entry_date = parse_date(entry["businessDate"])
|
||||
if entry_date < start_date or entry_date > end_date:
|
||||
continue
|
||||
matched_lines += 1
|
||||
bucket_start, bucket_end = bucket_for(entry["timestamp"], args.granularity)
|
||||
key = (
|
||||
iso_utc(bucket_start),
|
||||
iso_utc(bucket_end),
|
||||
entry["businessDate"],
|
||||
args.granularity,
|
||||
entry["domain"],
|
||||
entry["method"],
|
||||
entry["path"],
|
||||
)
|
||||
row = rows.setdefault(
|
||||
key,
|
||||
{
|
||||
"bucketStart": key[0],
|
||||
"bucketEnd": key[1],
|
||||
"businessDate": key[2],
|
||||
"granularity": key[3],
|
||||
"domain": key[4],
|
||||
"method": key[5],
|
||||
"path": key[6],
|
||||
"responseBytes": 0,
|
||||
"requestBytes": None,
|
||||
"requestBytesKnownCount": 0,
|
||||
"totalRequests": 0,
|
||||
"statusCounts": defaultdict(int),
|
||||
"sourceServer": args.source_server,
|
||||
"generatedAt": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
|
||||
"frozen": args.granularity == "day",
|
||||
},
|
||||
)
|
||||
row["responseBytes"] += entry["responseBytes"]
|
||||
row["totalRequests"] += 1
|
||||
row["statusCounts"][entry["statusBucket"]] += 1
|
||||
if entry["requestBytes"] is not None:
|
||||
row["requestBytes"] = (row["requestBytes"] or 0) + entry["requestBytes"]
|
||||
row["requestBytesKnownCount"] += 1
|
||||
|
||||
normalized_rows = []
|
||||
for row in rows.values():
|
||||
row["statusCounts"] = dict(row["statusCounts"])
|
||||
normalized_rows.append(row)
|
||||
|
||||
return normalized_rows, scanned_lines, matched_lines
|
||||
|
||||
|
||||
def post_rows(endpoint, token, rows, batch_size):
|
||||
accepted = 0
|
||||
for index in range(0, len(rows), batch_size):
|
||||
chunk = rows[index:index + batch_size]
|
||||
body = json.dumps({"rows": chunk}, ensure_ascii=False).encode("utf-8")
|
||||
request = urllib.request.Request(
|
||||
endpoint,
|
||||
data=body,
|
||||
method="POST",
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"X-Internal-Job-Token": token,
|
||||
},
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=240) as response:
|
||||
data = json.loads(response.read().decode("utf-8"))
|
||||
except urllib.error.HTTPError as exc:
|
||||
detail = exc.read().decode("utf-8", errors="replace")
|
||||
raise RuntimeError(f"POST failed HTTP {exc.code}: {detail}") from exc
|
||||
accepted += int(data.get("accepted", 0))
|
||||
return accepted
|
||||
|
||||
|
||||
def main():
|
||||
args = parse_args()
|
||||
token = os.environ.get(args.token_env, "").strip()
|
||||
if not token:
|
||||
raise SystemExit(f"Missing token env: {args.token_env}")
|
||||
|
||||
rows, scanned_lines, matched_lines = aggregate(args)
|
||||
if rows:
|
||||
accepted = post_rows(args.endpoint, token, rows, max(1, args.batch_size))
|
||||
else:
|
||||
accepted = 0
|
||||
|
||||
print(json.dumps({
|
||||
"ok": True,
|
||||
"granularity": args.granularity,
|
||||
"rows": len(rows),
|
||||
"accepted": accepted,
|
||||
"scannedLines": scanned_lines,
|
||||
"matchedLines": matched_lines,
|
||||
}, ensure_ascii=False))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Loading…
x
Reference in New Issue
Block a user