diff --git a/core/application/startup.go b/core/application/startup.go index a28f0f847..26b2f5b78 100644 --- a/core/application/startup.go +++ b/core/application/startup.go @@ -24,6 +24,7 @@ import ( "github.com/mudler/LocalAI/core/services/routing/router" "github.com/mudler/LocalAI/core/services/storage" coreStartup "github.com/mudler/LocalAI/core/startup" + "github.com/mudler/LocalAI/core/trace" "github.com/mudler/LocalAI/internal" "github.com/mudler/LocalAI/pkg/downloader" "github.com/mudler/LocalAI/pkg/modelartifacts" @@ -60,6 +61,7 @@ func New(opts ...config.AppOption) (*Application, error) { options.Threads = xsysinfo.CPUPhysicalCores() } + trace.ConfigureBackendTracePersistence(options.DataPath) application := newApplication(options) application.startupConfig = &startupConfigCopy @@ -133,7 +135,6 @@ func New(opts ...config.AppOption) (*Application, error) { migrateDataFiles(options.DynamicConfigsDir, options.DataPath) } } - // Initialize auth database if auth is enabled if options.Auth.Enabled { // Auto-generate HMAC secret if not provided diff --git a/core/http/middleware/trace.go b/core/http/middleware/trace.go index 77180677a..b15ebcc9d 100644 --- a/core/http/middleware/trace.go +++ b/core/http/middleware/trace.go @@ -7,6 +7,7 @@ import ( "mime" "net" "net/http" + "path/filepath" "slices" "strconv" "sync" @@ -17,6 +18,7 @@ import ( "github.com/labstack/echo/v4" "github.com/mudler/LocalAI/core/application" "github.com/mudler/LocalAI/core/http/auth" + "github.com/mudler/LocalAI/core/trace/tracepersist" "github.com/mudler/xlog" ) @@ -59,33 +61,94 @@ type APIExchange struct { var traceBuffer *circularbuffer.Queue[APIExchange] var mu sync.Mutex -var logChan = make(chan APIExchange, 100) -var tracingMaxItems int +var logChan = make(chan traceCommand, 100) var traceIDSeq atomic.Uint64 +var traceConsumerOnce sync.Once +var traceStore *tracepersist.Store[APIExchange] +var traceStoreKey string + +type traceCommand struct { + exchange *APIExchange + store *tracepersist.Store[APIExchange] + clear chan error +} func nextTraceID() string { return strconv.FormatUint(traceIDSeq.Add(1), 10) } -var doInitializeTracing = sync.OnceFunc(func() { - maxItems := tracingMaxItems +func initializeTracing(dataPath string, maxItems int) { if maxItems <= 0 { maxItems = 100 } + key := filepath.Join(dataPath, strconv.Itoa(maxItems)) mu.Lock() + if traceBuffer != nil && traceStoreKey == key { + mu.Unlock() + return + } + + var store *tracepersist.Store[APIExchange] + var restored []APIExchange + if dataPath != "" { + var err error + store, err = tracepersist.New[APIExchange](filepath.Join(dataPath, "traces", "api"), maxItems) + if err != nil { + xlog.Warn("Failed to initialize API trace persistence", "error", err) + } else if restored, err = store.Load(); err != nil { + xlog.Warn("Failed to restore API traces", "error", err) + store = nil + } + } traceBuffer = circularbuffer.New[APIExchange](maxItems) + for _, exchange := range restored { + traceBuffer.Enqueue(exchange) + advanceTraceID(&traceIDSeq, exchange.ID) + } + traceStore = store + traceStoreKey = key mu.Unlock() - go func() { - for exchange := range logChan { - mu.Lock() - if traceBuffer != nil { - traceBuffer.Enqueue(exchange) + traceConsumerOnce.Do(func() { + go func() { + for command := range logChan { + if command.clear != nil { + mu.Lock() + if traceBuffer != nil { + traceBuffer.Clear() + } + mu.Unlock() + var err error + if command.store != nil { + err = command.store.Clear() + } + command.clear <- err + continue + } + exchange := *command.exchange + if command.store != nil { + if err := command.store.Append(exchange.ID, exchange); err != nil { + xlog.Warn("Failed to persist API trace", "error", err) + } + } + mu.Lock() + if traceBuffer != nil { + traceBuffer.Enqueue(exchange) + } + mu.Unlock() } - mu.Unlock() - } - }() -}) + }() + }) +} + +func advanceTraceID(seq *atomic.Uint64, id string) { + n, err := strconv.ParseUint(id, 10, 64) + if err != nil { + return + } + for current := seq.Load(); n > current && !seq.CompareAndSwap(current, n); current = seq.Load() { + } +} type bodyWriter struct { http.ResponseWriter @@ -145,11 +208,6 @@ func (w *bodyWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) { return nil, nil, http.ErrNotSupported } -func initializeTracing(maxItems int) { - tracingMaxItems = maxItems - doInitializeTracing() -} - // sensitiveTraceHeaders is the set of header names whose values must not // land in the in-memory trace buffer. Keys are canonical — http.Header // stores them that way, so range yields canonical keys directly. @@ -175,14 +233,13 @@ func redactSensitiveHeaders(h http.Header) http.Header { // TraceMiddleware intercepts and logs JSON API requests and responses func TraceMiddleware(app *application.Application) echo.MiddlewareFunc { + initializeTracing(app.ApplicationConfig().DataPath, app.ApplicationConfig().TracingMaxItems) return func(next echo.HandlerFunc) echo.HandlerFunc { return func(c echo.Context) error { if !app.ApplicationConfig().EnableTracing { return next(c) } - initializeTracing(app.ApplicationConfig().TracingMaxItems) - ct, _, _ := mime.ParseMediaType(c.Request().Header.Get("Content-Type")) if ct != "application/json" { return next(c) @@ -266,8 +323,11 @@ func TraceMiddleware(app *application.Application) echo.MiddlewareFunc { exchange.UserName = user.Name } + mu.Lock() + store := traceStore + mu.Unlock() select { - case logChan <- exchange: + case logChan <- traceCommand{exchange: &exchange, store: store}: default: xlog.Warn("Trace channel full, dropping trace") } @@ -348,6 +408,18 @@ func window[T any](s []T, offset, limit int) []T { // ClearTraces clears the in-memory logs func ClearTraces() { + mu.Lock() + store := traceStore + initialized := traceBuffer != nil + mu.Unlock() + if initialized { + done := make(chan error, 1) + logChan <- traceCommand{store: store, clear: done} + if err := <-done; err != nil { + xlog.Warn("Failed to clear persisted API traces", "error", err) + } + return + } mu.Lock() if traceBuffer != nil { traceBuffer.Clear() diff --git a/core/http/middleware/trace_persistence_test.go b/core/http/middleware/trace_persistence_test.go new file mode 100644 index 000000000..16e943093 --- /dev/null +++ b/core/http/middleware/trace_persistence_test.go @@ -0,0 +1,47 @@ +// SPDX-License-Identifier: MIT + +package middleware + +import ( + "path/filepath" + "strconv" + "time" + + "github.com/mudler/LocalAI/core/trace/tracepersist" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("API trace persistence", func() { + It("restores before request execution and advances the ID sequence", func() { + dataPath := GinkgoT().TempDir() + store, err := tracepersist.New[APIExchange](filepath.Join(dataPath, "traces", "api"), 4) + Expect(err).NotTo(HaveOccurred()) + Expect(store.Append("41", APIExchange{ID: "41", Timestamp: time.Now()})).To(Succeed()) + + initializeTracing(dataPath, 4) + + Expect(GetTraces()).To(ContainElement(HaveField("ID", "41"))) + id, err := strconv.ParseUint(nextTraceID(), 10, 64) + Expect(err).NotTo(HaveOccurred()) + Expect(id).To(BeNumerically(">", 41)) + }) + + It("serializes clear behind records already queued for persistence", func() { + dataPath := GinkgoT().TempDir() + initializeTracing(dataPath, 64) + for i := range 50 { + exchange := APIExchange{ID: strconv.Itoa(i + 1), Timestamp: time.Now()} + logChan <- traceCommand{exchange: &exchange, store: traceStore} + } + + ClearTraces() + + store, err := tracepersist.New[APIExchange](filepath.Join(dataPath, "traces", "api"), 64) + Expect(err).NotTo(HaveOccurred()) + records, err := store.Load() + Expect(err).NotTo(HaveOccurred()) + Expect(records).To(BeEmpty()) + Expect(GetTraces()).To(BeEmpty()) + }) +}) diff --git a/core/trace/backend_trace.go b/core/trace/backend_trace.go index 951351f8c..6dbf03ab6 100644 --- a/core/trace/backend_trace.go +++ b/core/trace/backend_trace.go @@ -5,6 +5,7 @@ import ( "fmt" "maps" "math" + "path/filepath" "slices" "strconv" "sync" @@ -13,6 +14,7 @@ import ( "github.com/emirpasic/gods/v2/queues/circularbuffer" "github.com/mudler/LocalAI/core/schema" + "github.com/mudler/LocalAI/core/trace/tracepersist" "github.com/mudler/xlog" ) @@ -72,13 +74,22 @@ type BackendTrace struct { const MaxTraceBodyBytes = 1 << 20 var ( - backendTraceBuffer *circularbuffer.Queue[*BackendTrace] - backendMu sync.Mutex - backendLogChan = make(chan *BackendTrace, 100) - backendInitOnce sync.Once - backendTraceIDSeq atomic.Uint64 + backendTraceBuffer *circularbuffer.Queue[*BackendTrace] + backendMu sync.Mutex + backendLogChan = make(chan backendTraceCommand, 100) + backendConsumerOnce sync.Once + backendTraceIDSeq atomic.Uint64 + backendTraceDataPath string + backendTraceStore *tracepersist.Store[BackendTrace] + backendTraceStoreKey string ) +type backendTraceCommand struct { + trace *BackendTrace + store *tracepersist.Store[BackendTrace] + clear chan error +} + // backendMaxBodyBytes caps each captured string value in a BackendTrace.Data // field to keep the /api/backend-traces JSON small enough for the admin UI to // load on every 5s auto-refresh. Mirrors the API-trace body cap added in @@ -91,41 +102,86 @@ var ( var backendMaxBodyBytes int func InitBackendTracingIfEnabled(maxItems, maxBodyBytes int) { - backendInitOnce.Do(func() { - if maxItems <= 0 { - maxItems = 100 + if maxItems <= 0 { + maxItems = 100 + } + backendMu.Lock() + key := filepath.Join(backendTraceDataPath, strconv.Itoa(maxItems)) + if backendTraceBuffer == nil || backendTraceStoreKey != key { + var store *tracepersist.Store[BackendTrace] + var restored []BackendTrace + if backendTraceDataPath != "" { + var err error + store, err = tracepersist.New[BackendTrace](filepath.Join(backendTraceDataPath, "traces", "backend"), maxItems) + if err != nil { + xlog.Warn("Failed to initialize backend trace persistence", "error", err) + } else if restored, err = store.Load(); err != nil { + xlog.Warn("Failed to restore backend traces", "error", err) + store = nil + } } - backendMu.Lock() backendTraceBuffer = circularbuffer.New[*BackendTrace](maxItems) - backendMu.Unlock() + for i := range restored { + t := restored[i] + backendTraceBuffer.Enqueue(&t) + advanceBackendTraceID(t.ID) + } + backendTraceStore = store + backendTraceStoreKey = key + } + backendMaxBodyBytes = maxBodyBytes + backendMu.Unlock() + backendConsumerOnce.Do(func() { go func() { - for t := range backendLogChan { + for command := range backendLogChan { + if command.clear != nil { + backendMu.Lock() + if backendTraceBuffer != nil { + backendTraceBuffer.Clear() + } + backendMu.Unlock() + var err error + if command.store != nil { + err = command.store.Clear() + } + command.clear <- err + continue + } + if command.store != nil { + if err := command.store.Append(command.trace.ID, *command.trace); err != nil { + xlog.Warn("Failed to persist backend trace", "error", err) + } + } backendMu.Lock() if backendTraceBuffer != nil { - backendTraceBuffer.Enqueue(t) + backendTraceBuffer.Enqueue(command.trace) } backendMu.Unlock() } }() }) +} - // The body cap tracks the LATEST call, not the first: tracing_max_body_bytes - // is runtime-mutable via the settings API (ApplyRuntimeSettings), and every - // recording path calls this right before RecordBackendTrace with the current - // appConfig value. Freezing the cap on first init meant a raised setting let - // producers (e.g. trace.AudioSnippet, which reads the live value) embed - // payloads that this recorder then stomped with the "" - // marker — corrupting audio_wav_base64 into an unplayable string. maxItems - // keeps first-call semantics: resizing the ring buffer would drop entries. +func ConfigureBackendTracePersistence(dataPath string) { backendMu.Lock() - backendMaxBodyBytes = maxBodyBytes + backendTraceDataPath = dataPath backendMu.Unlock() } +func advanceBackendTraceID(id string) { + n, err := strconv.ParseUint(id, 10, 64) + if err != nil { + return + } + for current := backendTraceIDSeq.Load(); n > current && !backendTraceIDSeq.CompareAndSwap(current, n); current = backendTraceIDSeq.Load() { + } +} + func RecordBackendTrace(t BackendTrace) { backendMu.Lock() maxBody := backendMaxBodyBytes + store := backendTraceStore backendMu.Unlock() // Always walk Data, even with no body cap configured: besides capping // oversized strings (maxBody > 0), the walk replaces non-finite floats @@ -139,7 +195,7 @@ func RecordBackendTrace(t BackendTrace) { t.ID = strconv.FormatUint(backendTraceIDSeq.Add(1), 10) } select { - case backendLogChan <- &t: + case backendLogChan <- backendTraceCommand{trace: &t, store: store}: default: xlog.Warn("Backend trace channel full, dropping trace") } @@ -296,6 +352,28 @@ func SummarizeBackendTrace(t BackendTrace) BackendTrace { } func ClearBackendTraces() { + backendMu.Lock() + store := backendTraceStore + initialized := backendTraceBuffer != nil + dataPath := backendTraceDataPath + backendMu.Unlock() + if initialized { + done := make(chan error, 1) + backendLogChan <- backendTraceCommand{store: store, clear: done} + if err := <-done; err != nil { + xlog.Warn("Failed to clear persisted backend traces", "error", err) + } + return + } + if store == nil && dataPath != "" { + var err error + store, err = tracepersist.New[BackendTrace](filepath.Join(dataPath, "traces", "backend"), 100) + if err != nil { + xlog.Warn("Failed to initialize backend trace persistence for clear", "error", err) + } else if err := store.Clear(); err != nil { + xlog.Warn("Failed to clear persisted backend traces", "error", err) + } + } backendMu.Lock() if backendTraceBuffer != nil { backendTraceBuffer.Clear() diff --git a/core/trace/backend_trace_persistence_test.go b/core/trace/backend_trace_persistence_test.go new file mode 100644 index 000000000..dc24c27ac --- /dev/null +++ b/core/trace/backend_trace_persistence_test.go @@ -0,0 +1,55 @@ +// SPDX-License-Identifier: MIT + +package trace_test + +import ( + "path/filepath" + "strconv" + "time" + + "github.com/mudler/LocalAI/core/trace" + "github.com/mudler/LocalAI/core/trace/tracepersist" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("Backend trace persistence", func() { + It("restores traces and advances the ID sequence", func() { + dataPath := GinkgoT().TempDir() + store, err := tracepersist.New[trace.BackendTrace](filepath.Join(dataPath, "traces", "backend"), 4) + Expect(err).NotTo(HaveOccurred()) + Expect(store.Append("41", trace.BackendTrace{ + ID: "41", Timestamp: time.Now(), Type: trace.BackendTraceLLM, + })).To(Succeed()) + + trace.ConfigureBackendTracePersistence(dataPath) + trace.InitBackendTracingIfEnabled(4, 0) + + Eventually(trace.GetBackendTraces).Should(ContainElement(HaveField("ID", "41"))) + trace.RecordBackendTrace(trace.BackendTrace{Timestamp: time.Now(), Type: trace.BackendTraceLLM}) + Eventually(trace.GetBackendTraces).Should(HaveLen(2)) + id, err := strconv.ParseUint(trace.GetBackendTraces()[0].ID, 10, 64) + Expect(err).NotTo(HaveOccurred()) + Expect(id).To(BeNumerically(">", 41)) + }) + + It("serializes clear behind records already queued for persistence", func() { + dataPath := GinkgoT().TempDir() + trace.ConfigureBackendTracePersistence(dataPath) + trace.InitBackendTracingIfEnabled(64, 0) + + for i := range 50 { + trace.RecordBackendTrace(trace.BackendTrace{ + ID: strconv.Itoa(i + 1), Timestamp: time.Now(), Type: trace.BackendTraceLLM, + }) + } + trace.ClearBackendTraces() + + store, err := tracepersist.New[trace.BackendTrace](filepath.Join(dataPath, "traces", "backend"), 64) + Expect(err).NotTo(HaveOccurred()) + records, err := store.Load() + Expect(err).NotTo(HaveOccurred()) + Expect(records).To(BeEmpty()) + Expect(trace.GetBackendTraces()).To(BeEmpty()) + }) +}) diff --git a/core/trace/tracepersist/store.go b/core/trace/tracepersist/store.go new file mode 100644 index 000000000..72c00d041 --- /dev/null +++ b/core/trace/tracepersist/store.go @@ -0,0 +1,144 @@ +// SPDX-License-Identifier: MIT + +package tracepersist + +import ( + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "time" + + "github.com/mudler/xlog" +) + +type Store[T any] struct { + dir string + maxItems int + mu sync.Mutex + lastSeq int64 +} + +func New[T any](dir string, maxItems int) (*Store[T], error) { + if maxItems <= 0 { + maxItems = 100 + } + if err := os.MkdirAll(dir, 0o750); err != nil { + return nil, err + } + return &Store[T]{dir: dir, maxItems: maxItems}, nil +} + +func (s *Store[T]) Load() ([]T, error) { + s.mu.Lock() + defer s.mu.Unlock() + + files, err := s.files() + if err != nil { + return nil, err + } + records := make([]T, 0, len(files)) + for _, name := range files { + data, err := os.ReadFile(filepath.Join(s.dir, name)) + if err != nil { + return nil, err + } + var record T + if err := json.Unmarshal(data, &record); err != nil { + xlog.Warn("Skipping corrupt persisted trace", "file", name, "error", err) + continue + } + records = append(records, record) + } + return records, nil +} + +func (s *Store[T]) Append(id string, record T) error { + data, err := json.Marshal(record) + if err != nil { + return err + } + + s.mu.Lock() + defer s.mu.Unlock() + + seq := time.Now().UnixNano() + if seq <= s.lastSeq { + seq = s.lastSeq + 1 + } + s.lastSeq = seq + name := fmt.Sprintf("%020d-%s.json", seq, hex.EncodeToString([]byte(id))) + + tmp, err := os.CreateTemp(s.dir, ".trace-*.tmp") + if err != nil { + return err + } + tmpName := tmp.Name() + defer os.Remove(tmpName) + if err := tmp.Chmod(0o600); err != nil { + _ = tmp.Close() + return err + } + if _, err := tmp.Write(data); err != nil { + _ = tmp.Close() + return err + } + if err := tmp.Sync(); err != nil { + _ = tmp.Close() + return err + } + if err := tmp.Close(); err != nil { + return err + } + if err := os.Rename(tmpName, filepath.Join(s.dir, name)); err != nil { + return err + } + + files, err := s.files() + if err != nil { + return err + } + for len(files) > s.maxItems { + if err := os.Remove(filepath.Join(s.dir, files[0])); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + files = files[1:] + } + return nil +} + +func (s *Store[T]) Clear() error { + s.mu.Lock() + defer s.mu.Unlock() + + files, err := s.files() + if err != nil { + return err + } + for _, name := range files { + if err := os.Remove(filepath.Join(s.dir, name)); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + } + return nil +} + +func (s *Store[T]) files() ([]string, error) { + entries, err := os.ReadDir(s.dir) + if err != nil { + return nil, err + } + files := make([]string, 0, len(entries)) + for _, entry := range entries { + if !entry.IsDir() && strings.HasSuffix(entry.Name(), ".json") { + files = append(files, entry.Name()) + } + } + sort.Strings(files) + return files, nil +} diff --git a/core/trace/tracepersist/store_test.go b/core/trace/tracepersist/store_test.go new file mode 100644 index 000000000..6d9827e03 --- /dev/null +++ b/core/trace/tracepersist/store_test.go @@ -0,0 +1,73 @@ +// SPDX-License-Identifier: MIT + +package tracepersist_test + +import ( + "os" + "path/filepath" + "testing" + + "github.com/mudler/LocalAI/core/trace/tracepersist" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestTracePersist(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Trace persistence store suite") +} + +type record struct { + ID string `json:"id"` + Value string `json:"value"` +} + +var _ = Describe("Store", func() { + It("loads records oldest-first and skips corrupt files", func() { + dir := GinkgoT().TempDir() + store, err := tracepersist.New[record](dir, 3) + Expect(err).NotTo(HaveOccurred()) + + Expect(store.Append("1", record{ID: "1", Value: "first"})).To(Succeed()) + Expect(store.Append("2", record{ID: "2", Value: "second"})).To(Succeed()) + Expect(os.WriteFile(filepath.Join(dir, "00000000000000000003.json"), []byte("{"), 0o600)).To(Succeed()) + + records, err := store.Load() + Expect(err).NotTo(HaveOccurred()) + Expect(records).To(Equal([]record{ + {ID: "1", Value: "first"}, + {ID: "2", Value: "second"}, + })) + }) + + It("evicts the oldest records on disk", func() { + dir := GinkgoT().TempDir() + store, err := tracepersist.New[record](dir, 2) + Expect(err).NotTo(HaveOccurred()) + + Expect(store.Append("8", record{ID: "8"})).To(Succeed()) + Expect(store.Append("9", record{ID: "9"})).To(Succeed()) + Expect(store.Append("10", record{ID: "10"})).To(Succeed()) + + records, err := store.Load() + Expect(err).NotTo(HaveOccurred()) + Expect(records).To(Equal([]record{{ID: "9"}, {ID: "10"}})) + }) + + It("clears only files owned by the store", func() { + parent := GinkgoT().TempDir() + dir := filepath.Join(parent, "api") + store, err := tracepersist.New[record](dir, 2) + Expect(err).NotTo(HaveOccurred()) + Expect(store.Append("1", record{ID: "1"})).To(Succeed()) + other := filepath.Join(parent, "keep") + Expect(os.WriteFile(other, []byte("keep"), 0o600)).To(Succeed()) + + Expect(store.Clear()).To(Succeed()) + + records, err := store.Load() + Expect(err).NotTo(HaveOccurred()) + Expect(records).To(BeEmpty()) + Expect(other).To(BeAnExistingFile()) + }) +}) diff --git a/docs/content/features/tracing.md b/docs/content/features/tracing.md new file mode 100644 index 000000000..2fe25b349 --- /dev/null +++ b/docs/content/features/tracing.md @@ -0,0 +1,19 @@ ++++ +disableToc = false +title = "Tracing" +weight = 83 +url = '/features/tracing' ++++ + +LocalAI can retain recent API exchanges and backend operations for inspection +on the **Traces** page in the management interface. Enable tracing in runtime +settings or with the existing tracing configuration. + +API and backend trace histories are persisted in separate directories below +the configured data path. They are restored after a clean service restart, +whether or not authentication is enabled. + +Each history remains independently bounded by `tracing_max_items`. When a +history reaches that limit, LocalAI removes its oldest records from memory and +disk. The existing clear actions on the Traces page remove both the in-memory +history and its persisted records.