* feat(observability): OpenTelemetry logs+traces with secret redaction Part of #265. Adds opt-in OpenTelemetry (logs + traces) alongside the existing stderr + opslog pipeline, plus secret redaction on all sinks. Default-off: with no OTEL_* / SILO_OTEL_ENABLED config, behavior is unchanged. Bootstrap (internal/telemetry): - Setup() builds one shared resource, a TracerProvider (parent-based trace-id ratio sampler), a LoggerProvider, and the W3C TraceContext+Baggage propagator from env. It installs NO MeterProvider — metrics stay on Prometheus, and the built-in no-op global MeterProvider keeps the trace instrumentation libs from double-emitting. Shutdown is deferred with a flush timeout. - Logs are bridged via otelslog fan-out (slog.MultiHandler), level-gated by the shared LevelVar and best-effort so a failing collector can't break the console or DB branches. stderr + opslog stay untouched. Secret redaction (internal/logredact): - A slog.Handler masks secret-keyed attributes (password, token, api_key, authorization, cookie, ...) — including .With-bound attrs, nested groups, secret-keyed group subtrees, and values behind a LogValuer — on the console and OTLP sinks, with a no-op fast path when a record has no secret keys. opslog.shouldRedact delegates to logredact.SecretKey so all sinks share one marker list. Rotation is infra-managed (no custom file sink): container runtime for stderr, collector/backend for OTLP, opslog partition-pruning for the DB. Documented in docs/architecture/observability.md. Verification: go build ./..., go vet, gofmt -l — clean; go test ./internal/telemetry/ ./internal/logredact/ -race pass. AI-use disclosure: implemented with AI assistance (Claude Code), including adversarial reviews that hardened the bootstrap and fixed two redaction leak paths; reviewed by the author. * refactor(observability): slog context+component sweep, sloglint gate (phase 3) Part of #265. Builds on the OTel bootstrap + redaction commit. Standardizes every log call site onto the context-carrying slog variants so records correlate with the active OpenTelemetry trace, and locks the standard in with a machine gate so future code (human- or AI-authored) can't drift back. - Call-site sweep: converted the remaining slog.<Level>(...) calls to the slog.<Level>Context(ctx, ...) form wherever a context.Context is in scope (background/init calls with no ctx are left as-is), across 183 files. Applied via a type-aware AST codemod. Log levels and message strings are preserved verbatim; a component attr (canonical per-package name) is added to direct package-level slog calls. Bound-logger calls keep their existing .With bindings. The main.go and telemetry package conversions rode with their file in the previous commit to keep each file within a single commit. - Enforcement (.golangci.yml): enable sloglint with context=scope, static-msg, key-naming-case=snake, no-mixed-args. After the sweep all four report zero violations repo-wide (tests included), so make lint / CI now blocks any regression to the non-context form. The gate ships with the sweep because it cannot be green until the legacy sites are converted. Metrics remain on Prometheus; no behavior change to /metrics or Grafana. Verification: go build ./..., go vet ./..., gofmt -l — clean; sloglint (all 4 rules) 0 violations repo-wide; log levels verified unchanged. AI-use disclosure: implemented with AI assistance (Claude Code), including the codemod; reviewed by the author. * fix(observability): honor per-signal OTLP protocol and secret WithGroup names Two Codex review findings on PR #290: - telemetry: OTEL_EXPORTER_OTLP_{TRACES,LOGS}_PROTOCOL now override the generic OTEL_EXPORTER_OTLP_PROTOCOL per signal, so mixed collector setups (e.g. HTTP logs + gRPC traces) build the right exporter. - logredact: entering a group whose name is secret-bearing (e.g. WithGroup("authorization")) now masks every leaf in that subtree, matching how slog.Group("authorization", ...) is masked as a whole. * fix(observability): address review feedback on telemetry bootstrap - Telemetry setup failure no longer kills boot: Setup returns usable no-op providers alongside the error and main logs and continues with telemetry disabled, honoring the best-effort contract. - Honor OTEL_TRACES_SAMPLER (always_on/off, traceidratio, parentbased_* variants); unsupported values fall back to parentbased_traceidratio. - Attach node identity as semconv service.instance.id instead of the non-semconv node.name. - Rename opslog retention-scope log attrs to target_component/target_level so they no longer collide with the canonical component routing key, and tag those lines with component=opslog. - Fix stale levelGated comment casing; use WarnContext in the telemetry shutdown defer; document the LogValuer double-resolve on the redaction slow path. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Quick <31828688+Quick104@users.noreply.github.com> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
451 lines
14 KiB
Go
451 lines
14 KiB
Go
package handlers
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io/fs"
|
|
"log/slog"
|
|
"net/http"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/go-chi/chi/v5"
|
|
|
|
apimw "github.com/Silo-Server/silo-server/internal/api/middleware"
|
|
"github.com/Silo-Server/silo-server/internal/catalog"
|
|
"github.com/Silo-Server/silo-server/internal/collections/templates"
|
|
"github.com/Silo-Server/silo-server/internal/collectionutil"
|
|
"github.com/Silo-Server/silo-server/internal/mdblist"
|
|
"github.com/Silo-Server/silo-server/internal/s3client"
|
|
"github.com/Silo-Server/silo-server/internal/usercollections"
|
|
"github.com/Silo-Server/silo-server/internal/userstore"
|
|
)
|
|
|
|
// UserCollectionImportHandler exposes the user-side template gallery and
|
|
// import + sync endpoints. Authorization rules (only the creator can sync) are
|
|
// enforced here; the underlying sync.Service is intentionally unauthenticated.
|
|
type UserCollectionImportHandler struct {
|
|
storeProvider userstore.UserStoreProvider
|
|
sync *usercollections.Service
|
|
scheduler *usercollections.Scheduler
|
|
registry *templates.Registry
|
|
mdblist *mdblist.Client
|
|
s3GP *s3client.Client
|
|
frontendFS fs.FS
|
|
presignTTL time.Duration
|
|
}
|
|
|
|
func NewUserCollectionImportHandler(
|
|
provider userstore.UserStoreProvider,
|
|
sync *usercollections.Service,
|
|
scheduler *usercollections.Scheduler,
|
|
registry *templates.Registry,
|
|
mdblistClient *mdblist.Client,
|
|
s3GP *s3client.Client,
|
|
frontendFS fs.FS,
|
|
presignTTL time.Duration,
|
|
) *UserCollectionImportHandler {
|
|
if registry == nil {
|
|
registry = templates.Default
|
|
}
|
|
return &UserCollectionImportHandler{
|
|
storeProvider: provider,
|
|
sync: sync,
|
|
scheduler: scheduler,
|
|
registry: registry,
|
|
mdblist: mdblistClient,
|
|
s3GP: s3GP,
|
|
frontendFS: frontendFS,
|
|
presignTTL: presignTTL,
|
|
}
|
|
}
|
|
|
|
func (h *UserCollectionImportHandler) HandleListTemplates(w http.ResponseWriter, r *http.Request) {
|
|
writeJSON(w, http.StatusOK, h.registry.Catalog())
|
|
}
|
|
|
|
type userImportSharedFields struct {
|
|
Title string `json:"title"`
|
|
Description string `json:"description"`
|
|
Limit *int `json:"limit,omitempty"`
|
|
SyncSchedule string `json:"sync_schedule"`
|
|
IsShared bool `json:"is_shared"`
|
|
PosterURL string `json:"poster_url"`
|
|
LibraryIDs []int `json:"library_ids,omitempty"`
|
|
DisplayQueryDefinition json.RawMessage `json:"display_query_definition,omitempty"`
|
|
}
|
|
|
|
type userImportMDBListRequest struct {
|
|
userImportSharedFields
|
|
URL string `json:"url"`
|
|
}
|
|
|
|
type userImportTMDBRequest struct {
|
|
userImportSharedFields
|
|
Preset string `json:"preset"`
|
|
MediaType string `json:"media_type"`
|
|
TimeWindow string `json:"time_window"`
|
|
}
|
|
|
|
type userImportTraktRequest struct {
|
|
userImportSharedFields
|
|
Preset string `json:"preset"`
|
|
MediaType string `json:"media_type"`
|
|
}
|
|
|
|
type userImportResponse struct {
|
|
Collection collectionResponse `json:"collection"`
|
|
Sync *usercollections.SyncResult `json:"sync,omitempty"`
|
|
}
|
|
|
|
func (h *UserCollectionImportHandler) HandleImportMDBList(w http.ResponseWriter, r *http.Request) {
|
|
var req userImportMDBListRequest
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body")
|
|
return
|
|
}
|
|
if strings.TrimSpace(req.Title) == "" || strings.TrimSpace(req.URL) == "" {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "title and url are required")
|
|
return
|
|
}
|
|
if !validateOptionalLimit(req.Limit, w) {
|
|
return
|
|
}
|
|
cfg := usercollections.SourceConfig{
|
|
Mode: usercollections.SourceModeMDBList,
|
|
URL: usercollections.NormalizeMDBListURL(req.URL),
|
|
Limit: req.Limit,
|
|
LibraryIDs: req.LibraryIDs,
|
|
}
|
|
h.createImportedCollection(w, r, "mdblist", cfg, req.userImportSharedFields)
|
|
}
|
|
|
|
func (h *UserCollectionImportHandler) HandleImportTMDB(w http.ResponseWriter, r *http.Request) {
|
|
var req userImportTMDBRequest
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body")
|
|
return
|
|
}
|
|
if strings.TrimSpace(req.Title) == "" {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "title is required")
|
|
return
|
|
}
|
|
preset, mediaType, timeWindow, err := normalizeTMDBPresetRequest(req.Preset, req.MediaType, req.TimeWindow)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
|
|
return
|
|
}
|
|
if !validateOptionalLimit(req.Limit, w) {
|
|
return
|
|
}
|
|
cfg := usercollections.SourceConfig{
|
|
Mode: usercollections.SourceModeTMDBPreset,
|
|
Preset: preset,
|
|
MediaType: mediaType,
|
|
TimeWindow: timeWindow,
|
|
Limit: req.Limit,
|
|
LibraryIDs: req.LibraryIDs,
|
|
}
|
|
h.createImportedCollection(w, r, "tmdb", cfg, req.userImportSharedFields)
|
|
}
|
|
|
|
func (h *UserCollectionImportHandler) HandleImportTrakt(w http.ResponseWriter, r *http.Request) {
|
|
var req userImportTraktRequest
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body")
|
|
return
|
|
}
|
|
if strings.TrimSpace(req.Title) == "" {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "title is required")
|
|
return
|
|
}
|
|
profileID := apimw.GetProfileID(r.Context())
|
|
preset, mediaType, normalizedProfileID, err := normalizeTraktPresetRequest(req.Preset, req.MediaType, profileID)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
|
|
return
|
|
}
|
|
if !validateOptionalLimit(req.Limit, w) {
|
|
return
|
|
}
|
|
cfg := usercollections.SourceConfig{
|
|
Mode: usercollections.SourceModeTraktPreset,
|
|
Provider: "trakt",
|
|
Preset: preset,
|
|
MediaType: mediaType,
|
|
ProfileID: normalizedProfileID,
|
|
Limit: req.Limit,
|
|
LibraryIDs: req.LibraryIDs,
|
|
}
|
|
h.createImportedCollection(w, r, "trakt", cfg, req.userImportSharedFields)
|
|
}
|
|
|
|
func (h *UserCollectionImportHandler) createImportedCollection(
|
|
w http.ResponseWriter,
|
|
r *http.Request,
|
|
collectionType string,
|
|
cfg usercollections.SourceConfig,
|
|
shared userImportSharedFields,
|
|
) {
|
|
userID := apimw.GetUserID(r.Context())
|
|
profileID := apimw.GetProfileID(r.Context())
|
|
|
|
store, err := h.storeProvider.ForUser(r.Context(), userID)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to access user store")
|
|
return
|
|
}
|
|
|
|
schedule, err := usercollections.ResolveSyncSchedule(shared.SyncSchedule)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
|
|
return
|
|
}
|
|
if !validateOptionalLibraryIDs(cfg.LibraryIDs, w) {
|
|
return
|
|
}
|
|
|
|
sourceConfigJSON, err := usercollections.MarshalSourceConfig(cfg)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to encode source config")
|
|
return
|
|
}
|
|
displayQueryDefinition, err := catalog.NormalizeDisplayQueryFragment(shared.DisplayQueryDefinition)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
|
|
return
|
|
}
|
|
|
|
collection, err := store.CreateCollection(r.Context(), userstore.CreateCollectionInput{
|
|
CreatorProfileID: profileID,
|
|
Name: strings.TrimSpace(shared.Title),
|
|
Description: strings.TrimSpace(shared.Description),
|
|
CollectionType: collectionType,
|
|
IsShared: shared.IsShared,
|
|
QueryDefinition: "{}",
|
|
SortConfig: "{}",
|
|
SourceURL: cfg.DisplayURL(),
|
|
SourceConfig: sourceConfigJSON,
|
|
SyncSchedule: schedule,
|
|
NextSyncAt: usercollections.InitialNextSyncAt(schedule),
|
|
DisplayQueryDefinition: displayQueryDefinition,
|
|
PosterURL: strings.TrimSpace(shared.PosterURL),
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to create collection")
|
|
return
|
|
}
|
|
|
|
if err := h.storeBundledTemplatePoster(r, store, collection, strings.TrimSpace(shared.PosterURL)); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to process template poster")
|
|
return
|
|
}
|
|
|
|
syncResult, updated, syncErr := h.sync.RunSync(r.Context(), store, collection)
|
|
if syncErr != nil {
|
|
// Persist failure state inline so the UI shows the error and the user
|
|
// can retry; the row is intentionally kept around for that retry path.
|
|
_ = store.UpdateCollectionSyncState(r.Context(), userstore.UpdateCollectionSyncStateInput{
|
|
ID: collection.ID,
|
|
Status: "failed",
|
|
Message: syncErr.Error(),
|
|
LastSyncAt: time.Now().UTC(),
|
|
NextSyncAt: usercollections.InitialNextSyncAt(schedule),
|
|
})
|
|
updated = collection
|
|
updated.LastSyncStatus = "failed"
|
|
updated.LastSyncMessage = syncErr.Error()
|
|
}
|
|
|
|
writeJSON(w, http.StatusCreated, userImportResponse{
|
|
Collection: h.toCollectionResponse(r, *updated),
|
|
Sync: syncResult,
|
|
})
|
|
}
|
|
|
|
func (h *UserCollectionImportHandler) storeBundledTemplatePoster(
|
|
r *http.Request,
|
|
store userstore.UserStore,
|
|
collection *userstore.Collection,
|
|
posterPath string,
|
|
) error {
|
|
if collection == nil {
|
|
return nil
|
|
}
|
|
storedPath, thumbhash, stored, err := storeBundledCollectionPosterIfS3Configured(
|
|
r.Context(),
|
|
h.s3GP,
|
|
h.frontendFS,
|
|
collection.ID,
|
|
userCollectionImagePrefix,
|
|
posterPath,
|
|
)
|
|
if err != nil || !stored {
|
|
if err != nil {
|
|
slog.WarnContext(r.Context(), "failed to store bundled user collection poster", "component", "api",
|
|
"collection_id", collection.ID,
|
|
"poster_path", posterPath,
|
|
"error", err,
|
|
)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
if err := store.UpdateCollection(r.Context(), userstore.UpdateCollectionInput{
|
|
ID: collection.ID,
|
|
RequestProfileID: collection.CreatorProfileID,
|
|
PosterURL: &storedPath,
|
|
PosterThumbhash: &thumbhash,
|
|
}); err != nil {
|
|
slog.WarnContext(r.Context(), "failed to persist bundled user collection poster", "component", "api",
|
|
"collection_id", collection.ID,
|
|
"poster_path", posterPath,
|
|
"stored_path", storedPath,
|
|
"error", err,
|
|
)
|
|
return nil
|
|
}
|
|
collection.PosterURL = storedPath
|
|
collection.PosterThumbhash = thumbhash
|
|
return nil
|
|
}
|
|
|
|
func (h *UserCollectionImportHandler) toCollectionResponse(r *http.Request, c userstore.Collection) collectionResponse {
|
|
resp := toCollectionResponse(c)
|
|
resp.PosterURL = h.presignCollectionPoster(r.Context(), c.PosterURL)
|
|
return resp
|
|
}
|
|
|
|
func (h *UserCollectionImportHandler) presignCollectionPoster(ctx context.Context, path string) string {
|
|
if path == "" {
|
|
return ""
|
|
}
|
|
if strings.HasPrefix(path, "http://") || strings.HasPrefix(path, "https://") {
|
|
return path
|
|
}
|
|
if strings.HasPrefix(path, "/") {
|
|
return path
|
|
}
|
|
if h.s3GP == nil {
|
|
return ""
|
|
}
|
|
ttl := h.presignTTL
|
|
if ttl <= 0 {
|
|
ttl = 4 * time.Hour
|
|
}
|
|
url, err := h.s3GP.PresignGetURL(ctx, h.s3GP.Bucket(), cardThumbnailPath(path), ttl)
|
|
if err != nil {
|
|
return ""
|
|
}
|
|
return url
|
|
}
|
|
|
|
func (h *UserCollectionImportHandler) HandleSync(w http.ResponseWriter, r *http.Request) {
|
|
collectionID := chi.URLParam(r, "id")
|
|
if collectionID == "" {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "Collection ID is required")
|
|
return
|
|
}
|
|
userID := apimw.GetUserID(r.Context())
|
|
profileID := apimw.GetProfileID(r.Context())
|
|
|
|
store, err := h.storeProvider.ForUser(r.Context(), userID)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to access user store")
|
|
return
|
|
}
|
|
collection, err := store.GetCollection(r.Context(), collectionID)
|
|
if err != nil {
|
|
writeError(w, http.StatusNotFound, "not_found", "Collection not found")
|
|
return
|
|
}
|
|
if collection.CreatorProfileID != profileID {
|
|
writeError(w, http.StatusForbidden, "forbidden", "Only the creator can sync this collection")
|
|
return
|
|
}
|
|
if h.scheduler != nil && h.scheduler.IsInFlight(collectionID) {
|
|
writeError(w, http.StatusConflict, "sync_in_flight", "A sync is already running for this collection")
|
|
return
|
|
}
|
|
|
|
result, _, err := h.sync.RunSync(r.Context(), store, collection)
|
|
if err != nil {
|
|
if errors.Is(err, usercollections.ErrSyncUnsupported) {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "This collection does not support sync")
|
|
return
|
|
}
|
|
writeError(w, http.StatusInternalServerError, "internal_error", fmt.Sprintf("Sync failed: %v", err))
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, result)
|
|
}
|
|
|
|
type mdblistDiscoveryResponse struct {
|
|
Configured bool `json:"configured"`
|
|
Lists []mdblist.ListSummary `json:"lists"`
|
|
}
|
|
|
|
// mdblistConfigured returns true when the discovery client is usable. When
|
|
// false it has already written a "not configured" 200 response so callers
|
|
// can simply early-return.
|
|
func (h *UserCollectionImportHandler) mdblistConfigured(w http.ResponseWriter) bool {
|
|
if h.mdblist != nil && h.mdblist.Configured() {
|
|
return true
|
|
}
|
|
writeJSON(w, http.StatusOK, mdblistDiscoveryResponse{Configured: false, Lists: []mdblist.ListSummary{}})
|
|
return false
|
|
}
|
|
|
|
func (h *UserCollectionImportHandler) HandleSearchMDBList(w http.ResponseWriter, r *http.Request) {
|
|
if !h.mdblistConfigured(w) {
|
|
return
|
|
}
|
|
query := strings.TrimSpace(r.URL.Query().Get("q"))
|
|
if query == "" {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "q is required")
|
|
return
|
|
}
|
|
lists, err := h.mdblist.Search(r.Context(), query)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadGateway, "upstream_error", fmt.Sprintf("MDBList search failed: %v", err))
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, mdblistDiscoveryResponse{Configured: true, Lists: lists})
|
|
}
|
|
|
|
func (h *UserCollectionImportHandler) HandleTopMDBList(w http.ResponseWriter, r *http.Request) {
|
|
if !h.mdblistConfigured(w) {
|
|
return
|
|
}
|
|
lists, err := h.mdblist.Top(r.Context())
|
|
if err != nil {
|
|
writeError(w, http.StatusBadGateway, "upstream_error", fmt.Sprintf("MDBList top failed: %v", err))
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, mdblistDiscoveryResponse{Configured: true, Lists: lists})
|
|
}
|
|
|
|
func validateOptionalLimit(limit *int, w http.ResponseWriter) bool {
|
|
if limit == nil {
|
|
return true
|
|
}
|
|
if *limit <= 0 || *limit > collectionutil.MaxExplicitItemLimit {
|
|
writeError(w, http.StatusBadRequest, "bad_request",
|
|
fmt.Sprintf("limit must be between 1 and %d", collectionutil.MaxExplicitItemLimit))
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
func validateOptionalLibraryIDs(libraryIDs []int, w http.ResponseWriter) bool {
|
|
for _, id := range libraryIDs {
|
|
if id <= 0 {
|
|
writeError(w, http.StatusBadRequest, "bad_request", "library_ids must contain positive IDs")
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|