diff --git a/config/config.go b/config/config.go index 8d8d7e9..45602ff 100644 --- a/config/config.go +++ b/config/config.go @@ -80,6 +80,9 @@ func ReadAndInitConfig(configDir, configName, configType string) (eventRTConfig panic(err) } + config.BindEnv("mongodb.password", "MONGODB_PASSWORD") + config.BindEnv("service.secret_key", "SERVICE_SECRET_KEY") + if err := config.Unmarshal(&eventRTConfig); err != nil { panic(fmt.Sprintf("unmarshal eventRT config failed:%s\n", err.Error())) } diff --git a/deploy/mq/deploy.md b/deploy/deploy.md similarity index 100% rename from deploy/mq/deploy.md rename to deploy/deploy.md diff --git a/deploy/dockerfile/eventrt.Dockerfile b/deploy/dockerfile/eventrt.Dockerfile new file mode 100644 index 0000000..d18183e --- /dev/null +++ b/deploy/dockerfile/eventrt.Dockerfile @@ -0,0 +1,35 @@ +FROM golang:1.25-alpine AS builder +RUN apk --no-cache upgrade + +WORKDIR /app +COPY go.mod go.sum ./ +RUN GOPROXY="https://goproxy.cn,direct" go mod download +COPY . . +RUN CGO_ENABLED=0 GOOS=linux go build \ + -ldflags="-s -w" \ + -trimpath \ + -mod=readonly \ + -o eventrt main.go + +# Prepare runtime dependencies in a pinned Alpine stage so they can be +# copied into scratch without pulling any vulnerable OS packages at run time. +FROM alpine:3.21 AS certs +ARG USER_ID=1000 +RUN apk --no-cache add ca-certificates tzdata && \ + adduser -D -u ${USER_ID} eventrt + +FROM scratch +# CA certificates required for TLS connections (RabbitMQ amqps://) +COPY --from=certs /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ +# Timezone data +COPY --from=certs /usr/share/zoneinfo /usr/share/zoneinfo +# Non-root user/group definitions +COPY --from=certs /etc/passwd /etc/passwd +COPY --from=certs /etc/group /etc/group + +WORKDIR /app +COPY --from=builder /app/eventrt ./eventrt +COPY configs/config.example.yaml ./configs/config.example.yaml + +USER eventrt +CMD ["/app/eventrt", "-eventRT_config_dir=/app/configs"] diff --git a/deploy/k8s/eventrt-configmap.yaml b/deploy/k8s/eventrt-configmap.yaml new file mode 100644 index 0000000..a027848 --- /dev/null +++ b/deploy/k8s/eventrt-configmap.yaml @@ -0,0 +1,45 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: eventrt-config +data: + config.yaml: | + rabbitmq: + ca_cert_path: "/app/configs/certs/ca_certificate.pem" + client_key_path: "/app/configs/certs/eventrt_client_key.pem" + client_key_password: "" + client_cert_path: "/app/configs/certs/eventrt_client_cert.pem" + insecure_skip_verify: false + server_name: "rabbitmq-server" + user: "" + password: "" + host: "rabbitmq-service" + port: 5671 + + mongodb: + host: "192.168.1.101" + port: 27017 + user: "coslight" + password: "" # injected via env MONGODB_PASSWORD + database: "eventdb" + auth_db: "admin" + timeout: 10 + + logger: + mode: "production" + level: "info" + filepath: "" + maxsize: 100 + maxbackups: 5 + maxage: 30 + compress: false + + service: + service_addr: ":8081" + service_name: "eventRT" + secret_key: "" # injected via env SERVICE_SECRET_KEY + deploy_env: "production" + + otel: + endpoint: "jaeger:4318" + insecure: true diff --git a/deploy/k8s/eventrt-deployment.yaml b/deploy/k8s/eventrt-deployment.yaml new file mode 100644 index 0000000..b8e6091 --- /dev/null +++ b/deploy/k8s/eventrt-deployment.yaml @@ -0,0 +1,90 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: eventrt + labels: + app: eventrt +spec: + replicas: 1 + selector: + matchLabels: + app: eventrt + template: + metadata: + labels: + app: eventrt + spec: + containers: + - name: eventrt + image: coslight/eventrt:latest + imagePullPolicy: IfNotPresent + args: + - "-eventRT_config_dir=/app/configs" + - "-eventRT_config_name=config" + - "-eventRT_config_type=yaml" + ports: + - containerPort: 8081 + env: + # Downward API — injected into every log line by logger + - name: K8S_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace + - name: K8S_NODE_NAME + valueFrom: + fieldRef: + fieldPath: spec.nodeName + # HOSTNAME is set automatically by K8s to the pod name + # Sensitive values injected from Secret so they stay out of ConfigMap + - name: MONGODB_PASSWORD + valueFrom: + secretKeyRef: + name: eventrt-secret + key: mongodb-password + - name: SERVICE_SECRET_KEY + valueFrom: + secretKeyRef: + name: eventrt-secret + key: secret-key + volumeMounts: + - name: config + mountPath: /app/configs/config.yaml + subPath: config.yaml + readOnly: true + - name: certs + mountPath: /app/configs/certs + readOnly: true + resources: + requests: + cpu: 100m + memory: 128Mi + limits: + cpu: 500m + memory: 512Mi + securityContext: + runAsUser: 1000 + runAsNonRoot: true + readOnlyRootFilesystem: true + allowPrivilegeEscalation: false + capabilities: + drop: + - ALL + livenessProbe: + tcpSocket: + port: 8081 + initialDelaySeconds: 10 + periodSeconds: 30 + failureThreshold: 3 + readinessProbe: + tcpSocket: + port: 8081 + initialDelaySeconds: 5 + periodSeconds: 10 + failureThreshold: 3 + volumes: + - name: config + configMap: + name: eventrt-config + - name: certs + secret: + secretName: eventrt-certs diff --git a/deploy/k8s/eventrt-secret.yaml b/deploy/k8s/eventrt-secret.yaml new file mode 100644 index 0000000..b96df37 --- /dev/null +++ b/deploy/k8s/eventrt-secret.yaml @@ -0,0 +1,8 @@ +apiVersion: v1 +kind: Secret +metadata: + name: eventrt-secret +type: Opaque +stringData: + mongodb-password: "coslight@tj" + secret-key: "eventrt_key" diff --git a/deploy/k8s/eventrt-service.yaml b/deploy/k8s/eventrt-service.yaml new file mode 100644 index 0000000..25c0454 --- /dev/null +++ b/deploy/k8s/eventrt-service.yaml @@ -0,0 +1,15 @@ +apiVersion: v1 +kind: Service +metadata: + name: eventrt-service + labels: + app: eventrt +spec: + type: NodePort + selector: + app: eventrt + ports: + - name: http + port: 8081 + targetPort: 8081 + nodePort: 30081 diff --git a/deploy/mq/modelrt.cnf b/deploy/mq/modelrt.cnf deleted file mode 100644 index 2e4ada9..0000000 --- a/deploy/mq/modelrt.cnf +++ /dev/null @@ -1,14 +0,0 @@ -[req] -distinguished_name = req_distinguished_name -prompt = no - -[req_distinguished_name] -C = CN -ST = Beijing -L = Beijing -O = coslight -CN = modelrt-client - -[v3_client] -keyUsage = critical, digitalSignature, keyEncipherment -extendedKeyUsage = clientAuth diff --git a/docs/event-flow-analysis.md b/docs/event-flow-analysis.md new file mode 100644 index 0000000..ab276ea --- /dev/null +++ b/docs/event-flow-analysis.md @@ -0,0 +1,141 @@ +# eventRT 事件流转与存储分析 + +## 一、整体架构 + +```bash +modelRT 服务 + │ + │ 发布事件 (AMQPS + mTLS) + ▼ +RabbitMQ (exchange: event-exchange) + │ routing key: event.#.updown.* + ▼ +Queue: event-up-down-queue + │ + ▼ +eventRT (消费者) + │ + │ 解析 + 写入 + ▼ +MongoDB (eventdb.alarms) +``` + +--- + +## 二、事件数据结构 (`event/event.go`) + +| 字段 | 类型 | 说明 | +| :--- | :--- | :--- | +| `event` | string | 事件名称 | +| `event_uuid` | string | 唯一标识符 | +| `type` | int | 事件类型 | +| `priority` | int | 优先级 0-9 | +| `status` | int | 事件状态(见第四节) | +| `category` | string | 可选模板参数 | +| `timestamp` | int64 | 毫秒级 Unix 时间戳 | +| `from` | string | 来源:station / platform / msa | +| `condition` | map | 触发条件(如阈值、当前值) | +| `attached_subscriptions` | []any | 关联订阅信息 | +| `result` | map | 事件分析结果 | +| `operations` | []OperationRecord | 操作历史(见第四节) | +| `origin` | map | 子站原始告警数据 (CIM Alarm) | + +`OperationRecord` 结构: + +```go +Action string // 动作类型,如 "acknowledgment" +Op string // 操作人标识 +TS int64 // 操作时间(毫秒) +``` + +--- + +## 三、事件流转流程 (`event/up_down_limit_alarm.go`) + +```bash +main.go 启动 + └─ go event.ReceiptUpDownLimitAlarm(ctx) // goroutine 异步运行 + +ReceiptUpDownLimitAlarm(): + 1. GetConn() // 获取全局 RabbitMQ 连接 + 2. conn.Channel() // 创建 Channel + 3. channel.QueueBind( // 绑定路由 + queue = "event-up-down-queue", + exchange = "event-exchange", + key = "event.#.updown.*" + ) + 4. channel.Qos(1, 0, false) // 每次只预取 1 条,防止积压 + 5. channel.Consume(autoAck=false) // 手动 ACK 模式 + 6. for { select msg / ctx.Done() } // 事件循环 + +processAlarmEventMessage(msg): + 1. json.Unmarshal → EventRecord // 结构体反序列化 + 2. InsertOne(ctx, alarmEvent) // ⚠️ 第一次写入(结构体方式) + 3. bson.UnmarshalExtJSON → bson.D // 原始 BSON 反序列化 + 4. InsertOne(ctx, doc) // 第二次写入(原始文档方式) + 5. msg.Ack(false) // 手动确认消息 +``` + +--- + +## 四、事件状态流转 + +**现状:`Status` 字段有定义,但无状态常量,无状态机逻辑。** + +从 `EventRecord.Status int` + `Operations []OperationRecord` 的设计意图推断,其预期的状态流转应为: + +```bash +[未定义初始态] + │ + │ 由 modelRT 发布事件写入 + ▼ + status = ? (活跃/未确认) + │ + │ Operations 中追加 {action: "acknowledgment", op: 用户, ts: 时间} + ▼ + status = ? (已确认) + │ + │ 后续处理(result 填充等) + ▼ + status = ? (已关闭/已解决) +``` + +目前**状态码值没有任何枚举常量定义**,`constants/` 包下只有 trace、log_mode、deploy_mode,状态机逻辑尚未实现(代码有多处 `TODO`)。 + +--- + +## 五、存储方式 + +- **数据库**:MongoDB +- **库/集合**:`eventdb` / `alarms` +- **写入时机**:消息到达后立即插入,无缓冲 + +**当前代码存在一个明显问题**(`up_down_limit_alarm.go:99-115`): + +```go +// 第一次:以结构体插入(会被 Go 零值污染,如 omitempty 未设置的字段) +mongodbClient.Database("eventdb").Collection("alarms").InsertOne(ctx, alarmEvent) + +// 第二次:以原始 BSON 插入(保留原始字段,是更合理的方式) +mongodbClient.Database(dbName).Collection(collectionName).InsertOne(ctx, doc) +``` + +每条消息会**插入两份文档**到 `eventdb.alarms`,且 `msg.Ack` 只在第二次成功后才调用。这是个明显的 TODO 遗留问题,第一次插入应当被删除。 + +--- + +## 六、连接可靠性设计 + +RabbitMQ 连接有自动重连机制(`mq/mq_init.go:82`):断连后每 5 秒重试一次,直到 context 取消。MongoDB 目前无重连逻辑,直接依赖驱动内置能力。 + +--- + +## 七、总结 + +| 维度 | 现状 | +| :--- | :--- | +| 事件流转 | modelRT → RabbitMQ topic exchange → eventRT 消费 → MongoDB | +| 消息可靠性 | 手动 ACK + QoS=1 + 断连重连 | +| 存储方式 | MongoDB `eventdb.alarms`,目前每条消息重复写入两次(bug) | +| 状态流转 | `Status` + `Operations` 字段已建模,但状态常量和状态机**尚未实现** | +| 主要 TODO | 删除重复插入、定义状态枚举、实现状态流转逻辑、补充推送前端逻辑 | diff --git a/go.mod b/go.mod index 5b45c2f..ab36524 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module eventRT -go 1.25.0 +go 1.26.3 require ( github.com/gin-gonic/gin v1.11.0