Golang实现REST API通用操作日志与复制系统插件库oplog的使用

操作日志(OpLog)是记录系统变更的重要组件，在REST API开发中尤为关键。下面我将介绍如何使用Golang实现通用的操作日志系统，并介绍MongoDB oplog的集成使用。

一、基础操作日志实现

1. 定义操作日志模型

package models

import (
	"time"
)

type OperationLog struct {
	ID          string      `json:"id" bson:"_id"`
	UserID      string      `json:"user_id" bson:"user_id"`
	Username    string      `json:"username" bson:"username"`
	Operation   string      `json:"operation" bson:"operation"` // create/update/delete
	Resource    string      `json:"resource" bson:"resource"`   // 资源类型，如user/product
	ResourceID  string      `json:"resource_id" bson:"resource_id"`
	OldData     interface{} `json:"old_data,omitempty" bson:"old_data,omitempty"`
	NewData     interface{} `json:"new_data,omitempty" bson:"new_data,omitempty"`
	RequestData interface{} `json:"request_data,omitempty" bson:"request_data,omitempty"`
	IP          string      `json:"ip" bson:"ip"`
	CreatedAt   time.Time   `json:"created_at" bson:"created_at"`
}

2. 中间件记录操作日志

package middleware

import (
	"context"
	"net/http"
	"time"
	
	"github.com/yourproject/models"
)

func OpLogMiddleware(next http.Handler, logRepo models.OpLogRepository) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		// 获取请求信息
		userID := r.Header.Get("X-User-ID")
		username := r.Header.Get("X-Username")
		operation := getOperationFromMethod(r.Method)
		resource := getResourceFromPath(r.URL.Path)
		
		// 复制请求体以便后续使用
		var requestBody interface{}
		if r.Body != nil {
			// 这里简化处理，实际需要根据Content-Type解析
			// 可以使用io.TeeReader同时读取和复制请求体
		}
		
		// 包装ResponseWriter以捕获响应状态码和数据
		lrw := NewLoggingResponseWriter(w)
		
		// 调用下一个处理器
		next.ServeHTTP(lrw, r)
		
		// 记录日志
		log := models.OperationLog{
			UserID:      userID,
			Username:    username,
			Operation:   operation,
			Resource:    resource,
			RequestData: requestBody,
			IP:          r.RemoteAddr,
			CreatedAt:   time.Now(),
		}
		
		// 如果是修改操作，可以在这里获取变更前后的数据对比
		if operation == "update" || operation == "delete" {
			// 从上下文中获取旧数据(需要在业务处理中设置)
			if oldData := r.Context().Value("old_data"); oldData != nil {
				log.OldData = oldData
			}
		}
		
		// 保存日志
		go func() {
			_ = logRepo.Create(context.Background(), &log)
		}()
	})
}

type loggingResponseWriter struct {
	http.ResponseWriter
	statusCode int
}

func NewLoggingResponseWriter(w http.ResponseWriter) *loggingResponseWriter {
	return &loggingResponseWriter{w, http.StatusOK}
}

func (lrw *loggingResponseWriter) WriteHeader(code int) {
	lrw.statusCode = code
	lrw.ResponseWriter.WriteHeader(code)
}

二、集成MongoDB Oplog

MongoDB的oplog(操作日志)是一个特殊的集合，记录了所有对数据库的修改操作。

1. 监听MongoDB Oplog

package oplog

import (
	"context"
	"log"
	"time"
	
	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
)

type OplogWatcher struct {
	client     *mongo.Client
	dbName     string
	collection string
	callback   func(bson.Raw)
}

func NewOplogWatcher(mongoURI, dbName, collection string, callback func(bson.Raw)) (*OplogWatcher, error) {
	client, err := mongo.Connect(context.Background(), options.Client().ApplyURI(mongoURI))
	if err != nil {
		return nil, err
	}
	
	return &OplogWatcher{
		client:     client,
		dbName:     dbName,
		collection: collection,
		callback:   callback,
	}, nil
}

func (w *OplogWatcher) Watch() {
	// 获取oplog集合
	oplog := w.client.Database("local").Collection("oplog.rs")
	
	// 获取最后一条oplog的时间戳
	var lastEntry struct {
		Timestamp bson.MongoTimestamp `bson:"ts"`
	}
	opts := options.FindOne().SetSort(bson.D{{"$natural", -1}})
	err := oplog.FindOne(context.Background(), bson.D{}, opts).Decode(&lastEntry)
	if err != nil {
		log.Fatal(err)
	}
	
	// 创建监听管道
	pipeline := mongo.Pipeline{
		{{"$match", bson.D{
			{"ns", w.dbName + "." + w.collection},
			{"ts", bson.D{{"$gt", lastEntry.Timestamp}}},
		}}},
	}
	
	streamOpts := options.ChangeStream().SetFullDocument(options.UpdateLookup)
	stream, err := oplog.Watch(context.Background(), pipeline, streamOpts)
	if err != nil {
		log.Fatal(err)
	}
	
	// 处理变更流
	for stream.Next(context.Background()) {
		var changeEvent bson.Raw
		if err := stream.Decode(&changeEvent); err != nil {
			log.Println("Error decoding change event:", err)
			continue
		}
		w.callback(changeEvent)
	}
	
	if err := stream.Err(); err != nil {
		log.Println("Change stream error:", err)
	}
}

func (w *OplogWatcher) Close() {
	_ = w.client.Disconnect(context.Background())
}

2. 使用示例

func main() {
	// 初始化oplog监听器
	watcher, err := oplog.NewOplogWatcher(
		"mongodb://localhost:27017",
		"mydb",
		"users",
		func(change bson.Raw) {
			// 处理变更事件
			var event struct {
				Op string `bson:"op"` // i=insert, u=update, d=delete
				Ns string `bson:"ns"`
				O  bson.Raw
				O2 bson.Raw `bson:"o2,omitempty"`
			}
			
			if err := bson.Unmarshal(change, &event); err != nil {
				log.Println("Error unmarshalling change event:", err)
				return
			}
			
			log.Printf("Operation: %s on %s", event.Op, event.Ns)
			// 这里可以将变更转发到其他系统或进行其他处理
		},
	)
	if err != nil {
		log.Fatal(err)
	}
	defer watcher.Close()
	
	// 启动监听
	go watcher.Watch()
	
	// 启动HTTP服务器
	// ...
}

三、操作日志的应用场景

1. 审计追踪：记录谁在什么时候做了什么操作
2. 数据恢复：通过操作日志可以恢复误删或错误修改的数据
3. 数据同步：将变更同步到其他系统或数据库
4. 实时分析：分析用户行为模式

四、最佳实践

1. 异步记录：操作日志记录应该异步进行，不影响主业务流程
2. 批量写入：对于高并发系统，可以考虑批量写入日志
3. 日志轮转：定期归档旧日志，避免日志表过大
4. 敏感信息过滤：记录日志时过滤掉密码等敏感信息
5. 索引优化：为常用查询字段建立索引

通过以上实现，你可以构建一个完整的操作日志系统，既包含应用层的操作记录，又能利用数据库本身的变更日志机制，为你的REST API提供全面的操作追踪能力。