refactor(common): revert OSS ingest-log collection from #21711 (keep instrumentation only) (#21893)

This commit is contained in:
xugangqiang authored and GitHub committed 2026-11-03 15:50:26 +08:00
1 parent ebcdf2eef6
commit dd3c77c582
8 files changed
+72 -479

No files matched your search

+35
View File
@@ -0,0 +1,35 @@
//go:build !ee
package common_test
import (
"testing"
"ragflow/internal/common"
)
// TestSyncAuditNoPanic verifies the OSS no-op audit flush is callable and
// returns without panicking. The function must remain defined in the OSS stub
// so the server entrypoint (cmd/ragflow_server.go) compiles and runs.
func TestSyncAuditNoPanic(t *testing.T) {
defer func() {
if r := recover(); r != nil {
t.Fatalf("SyncAudit panicked: %v", r)
}
}()
common.SyncAudit()
}
// TestSyncAuditMultipleCalls guards against a regression where the no-op flush
// cannot be invoked repeatedly (e.g. an accidental removal that leaves a dangling
// deferred call in the server startup path).
func TestSyncAuditMultipleCalls(t *testing.T) {
defer func() {
if r := recover(); r != nil {
t.Fatalf("SyncAudit panicked on repeated call: %v", r)
}
}()
common.SyncAudit()
common.SyncAudit()
common.SyncAudit()
}
-68
View File
@@ -1,68 +0,0 @@
//go:build !ee
package common
import (
"os"
"path/filepath"
"testing"
"time"
)
// TestInitIngestLogPathTreatedAsFile is the regression test for P1b: the configured
// ingestor.log.path is a FILE name (per conf/service_conf.yaml comment), so InitIngestLog
// must write to that exact file and must NOT treat it as a directory.
func TestInitIngestLogPathTreatedAsFile(t *testing.T) {
t.Cleanup(resetIngestLog)
dir := t.TempDir()
filePath := filepath.Join(dir, "ingest_log.log")
if err := InitIngestLog("file", FileOutput{Path: filePath}, "api_server_9381"); err != nil {
t.Fatalf("InitIngestLog: %v", err)
}
LogIngestEvent(IngestionLog{EventTime: time.Now().UTC(), TaskID: "t1", EventType: "created", Status: "CREATED"})
SyncIngestLog()
fi, err := os.Stat(filePath)
if err != nil {
t.Fatalf("expected log file at %s: %v", filePath, err)
}
if fi.IsDir() {
t.Fatalf("%s was created as a DIRECTORY; path must be treated as a file", filePath)
}
if stray, err := os.Stat(filepath.Join(dir, "ingest_log.log")); err == nil && stray.IsDir() {
t.Fatalf("stray directory %s/ created; path was misinterpreted as a directory", dir)
}
}
// TestInitIngestLogEmptyPathNoFile asserts the documented default: an empty path emits to
// stdout only and opens no rotated file. This guards the off-by-default contract.
func TestInitIngestLogEmptyPathNoFile(t *testing.T) {
t.Cleanup(resetIngestLog)
dir := t.TempDir()
if err := InitIngestLog("file", FileOutput{}, "api_server_9381"); err != nil {
t.Fatalf("InitIngestLog: %v", err)
}
LogIngestEvent(IngestionLog{EventTime: time.Now().UTC(), TaskID: "t1"})
SyncIngestLog()
entries, _ := os.ReadDir(dir)
for _, e := range entries {
if nameContains(e.Name(), "ingest") {
t.Fatalf("empty path opened a file under %s: %s", dir, e.Name())
}
}
}
func nameContains(s, sub string) bool {
for i := 0; i+len(sub) <= len(s); i++ {
if s[i:i+len(sub)] == sub {
return true
}
}
return false
}
-186
View File
@@ -1,186 +0,0 @@
//go:build !ee
package common
import (
"bytes"
"encoding/json"
"os"
"path/filepath"
"reflect"
"strings"
"testing"
"time"
"go.uber.org/zap"
)
func resetIngestLog() {
ingestLogger.Store(zap.NewNop())
}
// TestIngestLogUnifiedStruct asserts the single unified type: renamed fields
// (DocID not DocumentID, EventType not TaskType) and the absence of the
// un-serializable Error error interface. Structural guard for 方案B unification.
func TestIngestLogUnifiedStruct(t *testing.T) {
typ := reflect.TypeOf(IngestionLog{})
for _, f := range []string{"DocID", "EventType", "ErrorInfo", "Code"} {
if _, ok := typ.FieldByName(f); !ok {
t.Errorf("IngestionLog missing expected unified field %q", f)
}
}
for _, bad := range []string{"DocumentID", "TaskType", "Error"} {
if _, ok := typ.FieldByName(bad); ok {
t.Errorf("IngestionLog must not keep legacy field %q after unification", bad)
}
}
}
// TestIngestLogEmitsUnifiedJSON happy path: flat JSON line with error_info (string) and code.
func TestIngestLogEmitsUnifiedJSON(t *testing.T) {
defer resetIngestLog()
dir := t.TempDir()
if err := InitIngestLog("json", FileOutput{Path: dir, Filename: "ingest.log"}, "svc"); err != nil {
t.Fatalf("InitIngestLog: %v", err)
}
LogIngestEvent(IngestionLog{
EventTime: time.Now().UTC(),
TenantID: "t1",
DatasetID: "kb1",
DocID: "d1",
UserID: "u1",
TaskID: "task1",
EventType: "completed",
Status: "COMPLETED",
ErrorInfo: "boom",
Filename: "report.pdf",
FileType: "pdf",
FileSize: 12345,
ParserID: "pdf",
Code: 7,
})
SyncIngestLog()
raw, err := os.ReadFile(filepath.Join(dir, "ingest.log"))
if err != nil {
t.Fatalf("read sink: %v", err)
}
var m map[string]any
if err := json.Unmarshal(raw, &m); err != nil {
t.Fatalf("emitted line not JSON: %v (raw=%q)", err, raw)
}
if m["error_info"] != "boom" {
t.Errorf("error_info = %v, want boom (string, not error interface)", m["error_info"])
}
if c, ok := m["code"].(float64); !ok || c != 7 {
t.Errorf("code = %v (%T), want 7", m["code"], m["code"])
}
if _, ok := m["error"]; ok {
t.Errorf("legacy error key must not be present; only error_info is allowed")
}
}
// TestIngestLogCodeRoundTrips adversarial: Code is a first-class struct field,
// not hard-coded 0 via a secondary mapping layer.
func TestIngestLogCodeRoundTrips(t *testing.T) {
defer resetIngestLog()
dir := t.TempDir()
if err := InitIngestLog("json", FileOutput{Path: dir, Filename: "ingest.log"}, "svc"); err != nil {
t.Fatalf("InitIngestLog: %v", err)
}
LogIngestEvent(IngestionLog{TaskID: "task1", EventType: "failed", Status: "FAILED", Code: 7})
SyncIngestLog()
raw, _ := os.ReadFile(filepath.Join(dir, "ingest.log"))
var m map[string]any
if err := json.Unmarshal(raw, &m); err != nil {
t.Fatalf("emitted line not JSON: %v (raw=%q)", err, raw)
}
if c, ok := m["code"].(float64); !ok || c != 7 {
t.Errorf("code = %v (%T), want 7 (first-class struct field, not hard-coded 0)", m["code"], m["code"])
}
}
// TestIngestLogOmitsZeroTimes adversarial: zero-value times must not leak.
func TestIngestLogOmitsZeroTimes(t *testing.T) {
defer resetIngestLog()
dir := t.TempDir()
if err := InitIngestLog("json", FileOutput{Path: dir, Filename: "ingest.log"}, "svc"); err != nil {
t.Fatalf("InitIngestLog: %v", err)
}
LogIngestEvent(IngestionLog{EventTime: time.Now().UTC(), TaskID: "t1", EventType: "created", Status: "CREATED"})
SyncIngestLog()
raw, _ := os.ReadFile(filepath.Join(dir, "ingest.log"))
if strings.Contains(string(raw), "0001-01-01") {
t.Errorf("emitted zero-time sentinel 0001-01-01: %s", raw)
}
var m map[string]any
json.Unmarshal(raw, &m)
for _, k := range []string{"begin_time", "end_time", "upload_time", "enqueue_time"} {
if _, ok := m[k]; ok {
t.Errorf("zero-time field %q must be omitted, got %v", k, m[k])
}
}
if _, ok := m["event_time"]; !ok {
t.Errorf("event_time must always be present")
}
}
// TestIngestLogOmitsEmptyErrorInfo adversarial: empty ErrorInfo omitted.
func TestIngestLogOmitsEmptyErrorInfo(t *testing.T) {
defer resetIngestLog()
dir := t.TempDir()
if err := InitIngestLog("json", FileOutput{Path: dir, Filename: "ingest.log"}, "svc"); err != nil {
t.Fatalf("InitIngestLog: %v", err)
}
LogIngestEvent(IngestionLog{TaskID: "t1", EventType: "created", Status: "CREATED"})
SyncIngestLog()
raw, _ := os.ReadFile(filepath.Join(dir, "ingest.log"))
var m map[string]any
json.Unmarshal(raw, &m)
if _, ok := m["error_info"]; ok {
t.Errorf("empty error_info must be omitted, got %v", m["error_info"])
}
}
// TestIngestLogOmitsEmptyUserName corner.
func TestIngestLogOmitsEmptyUserName(t *testing.T) {
defer resetIngestLog()
dir := t.TempDir()
if err := InitIngestLog("json", FileOutput{Path: dir, Filename: "ingest.log"}, "svc"); err != nil {
t.Fatalf("InitIngestLog: %v", err)
}
LogIngestEvent(IngestionLog{TaskID: "task1", EventType: "created", Status: "CREATED"})
SyncIngestLog()
raw, _ := os.ReadFile(filepath.Join(dir, "ingest.log"))
if strings.Contains(string(raw), "user_name") {
t.Fatalf("emitted line contains user_name when UserName is empty: %s", raw)
}
}
// TestIngestLogOneFlatLine adversarial: each event is exactly one JSON line.
func TestIngestLogOneFlatLine(t *testing.T) {
defer resetIngestLog()
dir := t.TempDir()
if err := InitIngestLog("json", FileOutput{Path: dir, Filename: "ingest.log"}, "svc"); err != nil {
t.Fatalf("InitIngestLog: %v", err)
}
LogIngestEvent(IngestionLog{TaskID: "task1", EventType: "completed", Status: "COMPLETED", ErrorInfo: "boom"})
SyncIngestLog()
raw, _ := os.ReadFile(filepath.Join(dir, "ingest.log"))
if len(bytes.TrimSpace(raw)) == 0 {
t.Fatal("LogIngest wrote no output")
}
lines := 0
for _, l := range bytes.Split(raw, []byte("\n")) {
if len(bytes.TrimSpace(l)) == 0 {
continue
}
lines++
var m map[string]any
if err := json.Unmarshal(l, &m); err != nil {
t.Fatalf("non-JSON line: %v (raw=%q)", err, l)
}
}
if lines != 1 {
t.Fatalf("expected 1 line, got %d (raw=%q)", lines, raw)
}
}
+34 -170
View File
@@ -12,22 +12,15 @@
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
//go:build !ee
package common
import (
"fmt"
"os"
"path/filepath"
"sync/atomic"
"time"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
"gopkg.in/natefinch/lumberjack.v2"
)
var (
@@ -61,182 +54,53 @@ type OperationLog struct {
func Audit(operationLog *OperationLog, fields ...zap.Field) {
}
// SyncAudit is the OSS no-op flush for the audit logger. The real sink lives in
// the EE build; the OSS binary keeps only the instrumentation seam so the
// server entrypoint's deferred flush (cmd/ragflow_server.go) compiles and runs.
func SyncAudit() {
}
// ingestLogger receives the serialized ingest events. It defaults to a no-op so a
// binary that never calls InitIngestLog emits nothing and never panics.
var ingestLogger atomic.Pointer[zap.Logger]
func init() {
ingestLogger.Store(zap.NewNop())
}
// InitIngestLog configures the sink for ingestion lifecycle events. It mirrors
// InitLogger: output always goes to stdout, and an optional rotated file is added
// when both Path and Filename are set.
// InitIngestLog is the OSS no-op hook for ingestion lifecycle event sinks. The OSS
// build does not collect these events; the real sink lives in the EE build via the
// LogIngest seam. The signature is retained so the IngestionEventEmitter /
// SetEventEmitter wiring in the ingestion service keeps compiling.
func InitIngestLog(logType string, file FileOutput, serviceName string) error {
encoderConfig := zapcore.EncoderConfig{
TimeKey: "timestamp",
LevelKey: "level",
NameKey: "service",
CallerKey: "caller",
FunctionKey: "",
MessageKey: "msg",
StacktraceKey: "stacktrace",
LineEnding: zapcore.DefaultLineEnding,
EncodeLevel: zapcore.LowercaseLevelEncoder,
EncodeTime: zapcore.TimeEncoderOfLayout("2006-01-02 15:04:05.000-07:00"),
EncodeDuration: zapcore.SecondsDurationEncoder,
EncodeCaller: zapcore.ShortCallerEncoder,
EncodeName: zapcore.FullNameEncoder,
}
maxSize := file.MaxSize
if maxSize <= 0 {
maxSize = DefaultLogMaxSizeMB
}
maxBackups := file.MaxBackups
if maxBackups <= 0 {
maxBackups = DefaultLogMaxBackups
}
maxAge := file.MaxAge
if maxAge <= 0 {
maxAge = DefaultLogMaxAgeDays
}
compress := DefaultLogCompress
if file.Compress != nil {
compress = *file.Compress
}
stdoutSyncer := zapcore.AddSync(os.Stdout)
syncers := []zapcore.WriteSyncer{stdoutSyncer}
// File output is opt-in via a non-empty Path. Path is treated as the log FILE
// (matching conf/service_conf.yaml's "path is the file" wording): a stray
// directory named after the file is never created. Only when Path is itself an
// existing directory do we synthesize a per-service file name inside it.
if file.Path != "" {
filename := file.Path
if fi, statErr := os.Stat(file.Path); statErr == nil && fi.IsDir() {
name := file.Filename
if name == "" {
name = fmt.Sprintf("%s_ingest.log", serviceName)
}
filename = filepath.Join(file.Path, name)
}
ljLogger := &lumberjack.Logger{
Filename: filename,
MaxSize: maxSize,
MaxBackups: maxBackups,
MaxAge: maxAge,
Compress: compress,
LocalTime: true,
}
syncers = append(syncers, zapcore.AddSync(ljLogger))
}
core := zapcore.NewCore(zapcore.NewJSONEncoder(encoderConfig), zapcore.NewMultiWriteSyncer(syncers...), zapcore.InfoLevel)
oldLogger := ingestLogger.Swap(zap.New(core))
if oldLogger != nil {
_ = oldLogger.Sync()
}
return nil
}
type IngestionLog struct {
EventTime time.Time `json:"event_time"`
TenantID string `json:"tenant_id"`
DatasetID string `json:"dataset_id"`
DocID string `json:"doc_id"`
UserID string `json:"user_id"`
UserName string `json:"user_name,omitempty"`
Worker string `json:"worker"`
TaskID string `json:"task_id"`
EventType string `json:"event_type"` // created|scheduled|running|completed|failed|stopped
Status string `json:"status"`
// ErrorInfo is the terminal reason as a plain string (not an error
// interface) so the event serializes cleanly and portably to external
// dashboards / ClickHouse.
ErrorInfo string `json:"error_info,omitempty"`
Filename string `json:"filename"`
FileType string `json:"file_type"`
FileSize int64 `json:"file_size"`
ParserID string `json:"parser_id"`
UploadTime time.Time `json:"upload_time"`
EnqueueTime time.Time `json:"enqueue_time"`
BeginTime time.Time `json:"begin_time"`
EndTime time.Time `json:"end_time"`
DurationMS int64 `json:"duration_ms"`
WaitDurationS float64 `json:"wait_duration_s"`
ParseDurationS float64 `json:"parse_duration_s"`
Progress float64 `json:"progress"`
RunCount int `json:"run_count"`
// Code is a machine-readable category for grouping in dashboards. It
// defaults to 0 when the caller does not set it.
Code int `json:"code"`
Operation map[string]any `json:"operation,omitempty"`
EventTime time.Time `json:"event_time"`
TenantID string `json:"tenant_id"`
DatasetID string `json:"dataset_id"`
DocID string `json:"doc_id"`
UserID string `json:"user_id"`
UserName string `json:"user_name,omitempty"`
Worker string `json:"worker"`
TaskID string `json:"task_id"`
EventType string `json:"event_type"`
Status string `json:"status"`
ErrorInfo string `json:"error_info,omitempty"`
Filename string `json:"filename"`
FileType string `json:"file_type"`
FileSize int64 `json:"file_size"`
ParserID string `json:"parser_id"`
UploadTime time.Time `json:"upload_time"`
EnqueueTime time.Time `json:"enqueue_time"`
BeginTime time.Time `json:"begin_time"`
EndTime time.Time `json:"end_time"`
DurationMS int64 `json:"duration_ms"`
WaitDurationS float64 `json:"wait_duration_s"`
ParseDurationS float64 `json:"parse_duration_s"`
Progress float64 `json:"progress"`
RunCount int `json:"run_count"`
Code int `json:"code"`
Operation map[string]any `json:"operation,omitempty"`
}
// SyncIngestLog flushes any buffered ingest output. Call it before process exit
// so no events are lost.
func SyncIngestLog() {
_ = ingestLogger.Load().Sync()
}
// LogIngestEvent transmits one structured ingestion lifecycle event as a single,
// flat JSON line. In the OSS build it writes to the local zap sink and also
// forwards to the shared LogIngest seam (a no-op in OSS, the real sink in EE).
func LogIngestEvent(log IngestionLog) {
ingestLogger.Load().With(log.zapFields()...).Info("ingest_log")
LogIngest(&log)
}
// zapFields renders the event as flat zap fields (top-level keys, UTC times) so
// the emitted JSON line is directly consumable by ClickHouse / Grafana.
func (l IngestionLog) zapFields() []zap.Field {
fields := []zap.Field{
zap.Time("event_time", l.EventTime.UTC()),
zap.String("tenant_id", l.TenantID),
zap.String("dataset_id", l.DatasetID),
zap.String("doc_id", l.DocID),
zap.String("user_id", l.UserID),
zap.String("worker", l.Worker),
zap.String("task_id", l.TaskID),
zap.String("event_type", l.EventType),
zap.String("status", l.Status),
zap.String("filename", l.Filename),
zap.String("file_type", l.FileType),
zap.Int64("file_size", l.FileSize),
zap.String("parser_id", l.ParserID),
skipZeroTime("upload_time", l.UploadTime),
skipZeroTime("enqueue_time", l.EnqueueTime),
skipZeroTime("begin_time", l.BeginTime),
skipZeroTime("end_time", l.EndTime),
zap.Int64("duration_ms", l.DurationMS),
zap.Float64("wait_duration_s", l.WaitDurationS),
zap.Float64("parse_duration_s", l.ParseDurationS),
zap.Float64("progress", l.Progress),
zap.Int("run_count", l.RunCount),
zap.Int("code", l.Code),
}
if l.UserName != "" {
fields = append(fields, zap.String("user_name", l.UserName))
}
if l.ErrorInfo != "" {
fields = append(fields, zap.String("error_info", l.ErrorInfo))
}
if len(l.Operation) > 0 {
fields = append(fields, zap.Any("operation", l.Operation))
}
return fields
}
// skipZeroTime omits a time field from the emitted JSON when it is the zero
// value, keeping the line compact and ClickHouse-friendly.
func skipZeroTime(key string, t time.Time) zap.Field {
if t.IsZero() {
return zap.Skip()
}
return zap.Time(key, t.UTC())
}
// LogIngest is the shared upstream seam. In the OSS build it is a no-op; the EE
// build provides the real sink. Our OSS LogIngestEvent forwards to it.
// build provides the real sink.
func LogIngest(log *IngestionLog, fields ...zap.Field) {}
+1 -1
View File
@@ -25,7 +25,7 @@ type CommonLogEmitter struct{}
// EmitIngestLog forwards the event to the common ingest-log sink.
func (e *CommonLogEmitter) EmitIngestLog(_ context.Context, log common.IngestionLog) {
common.LogIngestEvent(log)
common.LogIngest(&log)
}
// ComputeDurations derives wait (begin-enqueue) and parse (end-begin) seconds.
-1
View File
@@ -461,7 +461,6 @@ func startServer(ctx context.Context, serverName string, arguments *bootstrap.Se
// Initialize the ingestion lifecycle event sink (best-effort; stdout is
// always on, ingestor.log.path adds an optional rotated file). Flush it on shutdown.
bootstrap.SetIngestLog(serverName)
defer common.SyncIngestLog()
// Start server in a goroutine
go func() {
-2
View File
@@ -69,8 +69,6 @@ func SetLogger(serverName string, arguments *ServerArgs) {
func SetIngestLog(serverName string) {
globalConfig := server.GetConfig()
ingestLogConfig := globalConfig.GetIngestLogConfig()
// Flush any prior sink before swapping it out (e.g. on a reconfigure).
common.SyncIngestLog()
if err := common.InitIngestLog(ingestLogConfig.Type, ingestLogConfig.GetFileOutput(), serverName); err != nil {
common.Error("Failed to initialize ingest log sink", err)
}
+2 -51
View File
@@ -107,57 +107,8 @@ func (c *AuditConfig) GetFileOutput() common.FileOutput {
return common.FileOutput{}
}
// Default ingest-log rotation values mirror the documented IngestLogConfig
// defaults. File output is opt-in: leave ingestor.log.path empty and events emit
// to stdout only.
const (
defaultIngestLogMaxSizeMB = 1000
defaultIngestLogMaxBackups = 20
defaultIngestLogMaxAgeDays = 60
)
func (c *IngestLogConfig) ParseIngestLogConfig(v *viper.Viper) error { return nil }
// ParseIngestLogConfig reads the ingestor.log.* keys from the sub-viper passed by
// the ingestor config parser (already merged with the environment by
// server.Init), so env overrides such as RAGFLOW_INGESTOR_LOG_PATH win over the
// YAML file. A nil viper is a no-op so the struct keeps its zero (off) state.
func (c *IngestLogConfig) ParseIngestLogConfig(v *viper.Viper) error {
if v == nil {
return nil
}
c.Type = "file"
c.Format = "json"
c.MaxSize = defaultIngestLogMaxSizeMB
c.MaxBackups = defaultIngestLogMaxBackups
c.MaxAge = defaultIngestLogMaxAgeDays
if v.IsSet("type") {
c.Type = v.GetString("type")
}
if v.IsSet("format") {
c.Format = v.GetString("format")
}
if v.IsSet("path") {
c.Path = v.GetString("path")
}
if v.IsSet("max_size") {
c.MaxSize = v.GetInt("max_size")
}
if v.IsSet("max_backups") {
c.MaxBackups = v.GetInt("max_backups")
}
if v.IsSet("max_age") {
c.MaxAge = v.GetInt("max_age")
}
return nil
}
// GetFileOutput projects the ingest log config onto the common.FileOutput sink
// descriptor. An empty Path means no file sink (stdout only), which keeps the
// feature off by default.
func (c *IngestLogConfig) GetFileOutput() common.FileOutput {
return common.FileOutput{
Path: c.Path,
MaxSize: c.MaxSize,
MaxBackups: c.MaxBackups,
MaxAge: c.MaxAge,
}
return common.FileOutput{}
}