📈 技術調査 - Cloudflareが追加投資なしでセキュリティスキャンを10倍にした方法

package main

import (
    "context"
    "log"
    "sync"
    "time"

    "github.com/segmentio/kafka-go"
)

const (
    batchSize  = 100 // 1 バッチあたりの最大メッセージ数
    numWorkers = 16  // worker goroutine 数
)

func runConsumer(ctx context.Context, reader *kafka.Reader, processMsg func(kafka.Message) error) {
    for {
        batch := make([]kafka.Message, 0, batchSize)

        // 最初の 1 件はブロッキング取得 (自動コミットしない FetchMessage)
        msg, err := reader.FetchMessage(ctx)
        if err != nil {
            break
        }
        batch = append(batch, msg)

        // 残りは短いタイムアウトで詰め込む
    drainLoop:
        for len(batch) < batchSize {
            ctx2, cancel := context.WithTimeout(ctx, 5*time.Millisecond)
            m, err := reader.FetchMessage(ctx2)
            cancel()
            if err != nil {
                break drainLoop
            }
            batch = append(batch, m)
        }

        // バッチを worker pool で並列処理
        jobs := make(chan kafka.Message, len(batch))
        var wg sync.WaitGroup
        for i := 0; i < numWorkers; i++ {
            wg.Add(1)
            go func() {
                defer wg.Done()
                for m := range jobs {
                    if err := processMsg(m); err != nil {
                        log.Printf("process error: %v", err)
                    }
                }
            }()
        }
        for _, m := range batch {
            jobs <- m
        }
        close(jobs)
        wg.Wait()

        // 全 worker 完了後にバッチ末尾のオフセットをコミット
        if err := reader.CommitMessages(ctx, batch...); err != nil {
            log.Printf("commit error: %v", err)
        }
    }
}

// ルーター: メッセージを fast/slow topic に振り分ける
func routeMessage(ctx context.Context, msg kafka.Message, fastWriter, slowWriter *kafka.Writer) error {
    var payload ScanRequest
    if err := json.Unmarshal(msg.Value, &payload); err != nil {
        return err
    }

    // 判定: スキャン種別が重いかどうか
    dest := fastWriter
    if isHeavyScan(payload) {
        dest = slowWriter
    }

    return dest.WriteMessages(ctx, kafka.Message{
        Key:   msg.Key,
        Value: msg.Value,
    })
}

func isHeavyScan(req ScanRequest) bool {
    // 例: 外部 API を呼ぶチェッカーや処理時間が長いものを slow 判定
    heavyCheckers := map[string]bool{
        "ssl_full":     true,
        "third_party":  true,
        "deep_content": true,
    }
    return heavyCheckers[req.CheckerType]
}

// UNNEST バルク upsert (閾値以下)
func bulkUpsertWithUnnest(ctx context.Context, conn *pgx.Conn, issues []SecurityIssue) error {
    ids := make([]int64, len(issues))
    zoneIDs := make([]string, len(issues))
    severity := make([]string, len(issues))
    details := make([]string, len(issues))
    updatedAt := make([]time.Time, len(issues))

    for i, iss := range issues {
        ids[i] = iss.ID
        zoneIDs[i] = iss.ZoneID
        severity[i] = iss.Severity
        details[i] = iss.Details
        updatedAt[i] = iss.UpdatedAt
    }

    _, err := conn.Exec(ctx, `
        INSERT INTO security_issues (id, zone_id, severity, details, updated_at)
        SELECT * FROM UNNEST(
            $1::bigint[],
            $2::text[],
            $3::text[],
            $4::text[],
            $5::timestamptz[]
        )
        ON CONFLICT (id) DO UPDATE
            SET severity   = EXCLUDED.severity,
                details    = EXCLUDED.details,
                updated_at = EXCLUDED.updated_at
    `, ids, zoneIDs, severity, details, updatedAt)

    return err
}

// CopyFrom バルク insert (閾値超過時、ステージングテーブルへ)
// db は *pgx.Conn / pgxpool.Pool / pgx.Tx いずれも満たす (CopyFrom を持つ)
func bulkInsertWithCopy(ctx context.Context, db interface {
    CopyFrom(context.Context, pgx.Identifier, []string, pgx.CopyFromSource) (int64, error)
}, issues []SecurityIssue) error {
    _, err := db.CopyFrom(
        ctx,
        pgx.Identifier{"security_issues_staging"},
        []string{"id", "zone_id", "severity", "details", "updated_at"},
        pgx.CopyFromSlice(len(issues), func(i int) ([]any, error) {
            return []any{
                issues[i].ID,
                issues[i].ZoneID,
                issues[i].Severity,
                issues[i].Details,
                issues[i].UpdatedAt,
            }, nil
        }),
    )
    return err
}

// ハイブリッド切り替えロジック
const unnestThreshold = 2000 // この件数を超えたら COPY に切り替え

func bulkUpsert(ctx context.Context, conn *pgx.Conn, issues []SecurityIssue) error {
    if len(issues) <= unnestThreshold {
        return bulkUpsertWithUnnest(ctx, conn, issues)
    }
    return bulkInsertViaCopy(ctx, conn, issues)
}

func bulkInsertViaCopy(ctx context.Context, conn *pgx.Conn, issues []SecurityIssue) error {
    // staging への投入 → upsert を 1 トランザクションで囲む
    tx, err := conn.Begin(ctx)
    if err != nil {
        return err
    }
    defer tx.Rollback(ctx) // commit 済みなら no-op

    // 0. トランザクション内のみ有効な一時テーブルを作る (並行リクエストで競合しない)
    if _, err := tx.Exec(ctx, `
        CREATE TEMP TABLE security_issues_staging
        (LIKE security_issues INCLUDING DEFAULTS)
        ON COMMIT DROP
    `); err != nil {
        return err
    }
    // 1. COPY で一時テーブルに高速投入
    if err := bulkInsertWithCopy(ctx, tx, issues); err != nil {
        return err
    }
    // 2. 一時テーブル → 本テーブルへ upsert
    if _, err := tx.Exec(ctx, `
        INSERT INTO security_issues
        SELECT * FROM security_issues_staging
        ON CONFLICT (id) DO UPDATE
            SET severity   = EXCLUDED.severity,
                details    = EXCLUDED.details,
                updated_at = EXCLUDED.updated_at
    `); err != nil {
        return err
    }
    return tx.Commit(ctx)
}

package scanner

import (
    "context"
    "math"
    "time"

    "golang.org/x/time/rate"
)

// プランごとのスキャン間隔設定
const (
    freeScanInterval time.Duration = 7 * 24 * time.Hour
    proScanInterval  time.Duration = 3 * 24 * time.Hour
    bizScanInterval  time.Duration = 3 * 24 * time.Hour
    entScanInterval  time.Duration = 24 * time.Hour

    rateLimitBufferFactor = 1.1 // 10% のバッファを持たせる
    rateRecalcInterval    = 30 * time.Minute
)

// computeRate: 対象件数からレートを計算する (事例の手法を反映した実装例)
func computeRate(free, pro, biz, ent int64) rate.Limit {
    r := float64(free)/freeScanInterval.Seconds() +
        float64(pro)/proScanInterval.Seconds() +
        float64(biz)/bizScanInterval.Seconds() +
        float64(ent)/entScanInterval.Seconds()

    r *= rateLimitBufferFactor
    return rate.Limit(r)
}

type RateAdaptingScanner struct {
    limiter *rate.Limiter
    db      DB
}

func NewRateAdaptingScanner(db DB) *RateAdaptingScanner {
    s := &RateAdaptingScanner{
        limiter: rate.NewLimiter(1, 1), // 初期値: 仮設定
        db:      db,
    }
    go s.recalcLoop()
    return s
}

// recalcLoop: 30 分ごとにレートを再計算してリミッターを更新する
func (s *RateAdaptingScanner) recalcLoop() {
    ticker := time.NewTicker(rateRecalcInterval)
    defer ticker.Stop()

    for range ticker.C {
        counts, err := s.db.CountZonesByPlan(context.Background())
        if err != nil {
            continue
        }
        newLimit := computeRate(counts.Free, counts.Pro, counts.Biz, counts.Ent)
        s.limiter.SetLimit(newLimit)
        // burst は最低 1 を確保 (低レート時に Wait が永久ブロックしないように)
        burst := int(math.Ceil(float64(newLimit) * 2))
        if burst < 1 {
            burst = 1
        }
        s.limiter.SetBurst(burst)
    }
}

// Scan: レートリミッターに従ってスキャンを実行する
func (s *RateAdaptingScanner) Scan(ctx context.Context, zone Zone) error {
    if err := s.limiter.Wait(ctx); err != nil {
        return err
    }
    return performScan(zone)
}

package db

import (
    "context"
    "fmt"
    "os"

    "github.com/jackc/pgx/v5/pgxpool"
)

type DBCluster struct {
    primary *pgxpool.Pool // 書き込み用: プライマリ DB
    replica *pgxpool.Pool // 読み取り専用: リージョンローカルレプリカ
    region  string
}

func NewDBCluster(ctx context.Context, primaryDSN, replicaDSN string) (*DBCluster, error) {
    primary, err := pgxpool.New(ctx, primaryDSN)
    if err != nil {
        return nil, fmt.Errorf("primary pool: %w", err)
    }
    replica, err := pgxpool.New(ctx, replicaDSN)
    if err != nil {
        return nil, fmt.Errorf("replica pool: %w", err)
    }
    return &DBCluster{
        primary: primary,
        replica: replica,
        region:  os.Getenv("REGION"),
    }, nil
}

// WriteConn: 書き込みは常にプライマリへ
func (c *DBCluster) WriteConn() *pgxpool.Pool { return c.primary }

// ReadConn: 読み取りはリージョンローカルのレプリカへ
func (c *DBCluster) ReadConn() *pgxpool.Pool { return c.replica }

# 最遅パーティションのラグが閾値超過、かつ消費が止まっている状態を検知
max by(group, topic)(kafka_consumer_group_partition_lag) > 10000
  and rate(kafka_consumer_fetch_manager_records_consumed_total[5m]) == 0

# consumer lag をグループ単位で一覧
kafka-consumer-groups.sh \
  --bootstrap-server <broker>:9092 \
  --group scan-consumer \
  --describe

-- 設定間隔の 1.2 倍を超えて未スキャンのゾーンを確認
SELECT zone_id, last_scanned_at, scan_interval_minutes,
       EXTRACT(EPOCH FROM (NOW() - last_scanned_at))/60 AS lag_minutes
FROM zone_scan_status
WHERE EXTRACT(EPOCH FROM (NOW() - last_scanned_at))/60 > scan_interval_minutes * 1.2
ORDER BY lag_minutes DESC;

import random

def schedule_zone(zone_id, interval_seconds):
    jitter = random.uniform(0, interval_seconds)
    next_run = now() + jitter
    update_zone_schedule(zone_id, next_run)

required_rate = Σ(zone_count_per_plan / scan_interval_per_plan)
actual_rate_limit = required_rate × buffer_factor  # buffer_factor: 1.1〜1.3