mirrors/at-container-registry
1
0
Fork 0
mirror of https://tangled.org/evan.jarrett.net/at-container-registry synced 2026-09-30 14:05:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

add log shipper begin envvar cleanup

Browse Source
This commit is contained in:
Evan Jarrett
2026-01-08 22:52:32 -06:00
parent f27e2e0d93
commit 51f6917444
33 changed files with 1430 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/appview/config.go
+48 -47
View File
@@ -13,7 +13,6 @@ import (
"net/url"
"os"
"strconv"
"strings"
"time"
"github.com/distribution/distribution/v3/configuration"
@@ -23,6 +22,7 @@ import (
type Config struct {
Version string `yaml:"version"`
LogLevel string `yaml:"log_level"`
LogShipper LogShipperConfig `yaml:"log_shipper"`
Server ServerConfig `yaml:"server"`
UI UIConfig `yaml:"ui"`
Health HealthConfig `yaml:"health"`
@@ -32,6 +32,28 @@ type Config struct {
Distribution *configuration.Configuration `yaml:"-"` // Wrapped distribution config for compatibility
}
// LogShipperConfig defines remote log shipping settings
type LogShipperConfig struct {
// Backend selects the log shipping backend (from env: ATCR_LOG_SHIPPER_BACKEND)
// Valid values: "victoria", "opensearch", "loki", or empty to disable
Backend string `yaml:"backend"`
// URL is the remote log service endpoint (from env: ATCR_LOG_SHIPPER_URL)
URL string `yaml:"url"`
// BatchSize is the number of logs to batch before flushing (from env: ATCR_LOG_SHIPPER_BATCH_SIZE, default: 100)
BatchSize int `yaml:"batch_size"`
// FlushInterval is the max time between flushes (from env: ATCR_LOG_SHIPPER_FLUSH_INTERVAL, default: 5s)
FlushInterval time.Duration `yaml:"flush_interval"`
// Username for basic auth (from env: ATCR_LOG_SHIPPER_USERNAME, optional)
Username string `yaml:"username"`
// Password for basic auth (from env: ATCR_LOG_SHIPPER_PASSWORD, optional)
Password string `yaml:"password"`
}
// ServerConfig defines server settings
type ServerConfig struct {
// Addr is the HTTP listen address (from env: ATCR_HTTP_ADDR, default: ":5000")
@@ -48,9 +70,6 @@ type ServerConfig struct {
// TestMode enables HTTP for local DID resolution and transition:generic scope (from env: TEST_MODE)
TestMode bool `yaml:"test_mode"`
// DebugAddr is the debug/pprof HTTP listen address (from env: ATCR_DEBUG_ADDR, default: ":5001")
DebugAddr string `yaml:"debug_addr"`
// OAuthKeyPath is the path to the OAuth client P-256 signing key (from env: ATCR_OAUTH_KEY_PATH, default: "/var/lib/atcr/oauth/client.key")
// Auto-generated on first run for production (non-localhost) deployments
OAuthKeyPath string `yaml:"oauth_key_path"`
@@ -62,14 +81,8 @@ type ServerConfig struct {
// UIConfig defines web UI settings
type UIConfig struct {
// Enabled controls whether the web UI is enabled (from env: ATCR_UI_ENABLED, default: true)
Enabled bool `yaml:"enabled"`
// DatabasePath is the path to the UI SQLite database (from env: ATCR_UI_DATABASE_PATH, default: "/var/lib/atcr/ui.db")
DatabasePath string `yaml:"database_path"`
// SkipDBMigrations controls whether to skip running database migrations (from env: SKIP_DB_MIGRATIONS, default: false)
SkipDBMigrations bool `yaml:"skip_db_migrations"`
}
// HealthConfig defines health check and cache settings
@@ -112,19 +125,11 @@ type AuthConfig struct {
ServiceName string `yaml:"service_name"`
}
// CredentialHelperConfig defines credential helper version and download settings
// CredentialHelperConfig defines credential helper download settings
type CredentialHelperConfig struct {
// Version is the latest credential helper version (from env: ATCR_CREDENTIAL_HELPER_VERSION)
// e.g., "v0.0.2"
Version string `yaml:"version"`
// TangledRepo is the Tangled repository URL for downloads (from env: ATCR_CREDENTIAL_HELPER_TANGLED_REPO)
// Default: "https://tangled.org/@evan.jarrett.net/at-container-registry"
// TangledRepo is the Tangled repository URL for downloads
// Hardcoded default: "https://tangled.org/@evan.jarrett.net/at-container-registry"
TangledRepo string `yaml:"tangled_repo"`
// Checksums is a comma-separated list of platform:sha256 pairs (from env: ATCR_CREDENTIAL_HELPER_CHECKSUMS)
// e.g., "linux_amd64:abc123,darwin_arm64:def456"
Checksums map[string]string `yaml:"-"`
}
// LoadConfigFromEnv builds a complete configuration from environment variables
@@ -137,9 +142,16 @@ func LoadConfigFromEnv() (*Config, error) {
// Logging configuration
cfg.LogLevel = getEnvOrDefault("ATCR_LOG_LEVEL", "info")
// Log shipper configuration
cfg.LogShipper.Backend = os.Getenv("ATCR_LOG_SHIPPER_BACKEND")
cfg.LogShipper.URL = os.Getenv("ATCR_LOG_SHIPPER_URL")
cfg.LogShipper.BatchSize = getIntOrDefault("ATCR_LOG_SHIPPER_BATCH_SIZE", 100)
cfg.LogShipper.FlushInterval = getDurationOrDefault("ATCR_LOG_SHIPPER_FLUSH_INTERVAL", 5*time.Second)
cfg.LogShipper.Username = os.Getenv("ATCR_LOG_SHIPPER_USERNAME")
cfg.LogShipper.Password = os.Getenv("ATCR_LOG_SHIPPER_PASSWORD")
// Server configuration
cfg.Server.Addr = getEnvOrDefault("ATCR_HTTP_ADDR", ":5000")
cfg.Server.DebugAddr = getEnvOrDefault("ATCR_DEBUG_ADDR", ":5001")
cfg.Server.DefaultHoldDID = os.Getenv("ATCR_DEFAULT_HOLD_DID")
if cfg.Server.DefaultHoldDID == "" {
return nil, fmt.Errorf("ATCR_DEFAULT_HOLD_DID is required")
@@ -155,9 +167,7 @@ func LoadConfigFromEnv() (*Config, error) {
}
// UI configuration
cfg.UI.Enabled = os.Getenv("ATCR_UI_ENABLED") != "false"
cfg.UI.DatabasePath = getEnvOrDefault("ATCR_UI_DATABASE_PATH", "/var/lib/atcr/ui.db")
cfg.UI.SkipDBMigrations = os.Getenv("SKIP_DB_MIGRATIONS") == "true"
// Health and cache configuration
cfg.Health.CacheTTL = getDurationOrDefault("ATCR_HEALTH_CACHE_TTL", 15*time.Minute)
@@ -184,10 +194,8 @@ func LoadConfigFromEnv() (*Config, error) {
// Derive service name from base URL or env var (used for JWT issuer and service)
cfg.Auth.ServiceName = getServiceName(cfg.Server.BaseURL)
// Credential helper configuration
cfg.CredentialHelper.Version = os.Getenv("ATCR_CREDENTIAL_HELPER_VERSION")
cfg.CredentialHelper.TangledRepo = getEnvOrDefault("ATCR_CREDENTIAL_HELPER_TANGLED_REPO", "https://tangled.org/@evan.jarrett.net/at-container-registry")
cfg.CredentialHelper.Checksums = parseChecksums(os.Getenv("ATCR_CREDENTIAL_HELPER_CHECKSUMS"))
// Credential helper configuration (hardcoded - no env vars needed)
cfg.CredentialHelper.TangledRepo = "https://tangled.org/@evan.jarrett.net/at-container-registry"
// Build distribution configuration for compatibility with distribution library
distConfig, err := buildDistributionConfig(cfg)
@@ -233,9 +241,6 @@ func buildDistributionConfig(cfg *Config) (*configuration.Configuration, error)
Headers: map[string][]string{
"X-Content-Type-Options": {"nosniff"},
},
Debug: configuration.Debug{
Addr: cfg.Server.DebugAddr,
},
}
// Storage (fake in-memory placeholder - all real storage is proxied)
@@ -380,23 +385,19 @@ func getDurationOrDefault(envKey string, defaultValue time.Duration) time.Durati
return parsed
}
// parseChecksums parses a comma-separated list of platform:sha256 pairs
// e.g., "linux_amd64:abc123,darwin_arm64:def456"
func parseChecksums(checksumsStr string) map[string]string {
checksums := make(map[string]string)
if checksumsStr == "" {
return checksums
// getIntOrDefault parses an int from environment variable or returns default
// Logs a warning if parsing fails
func getIntOrDefault(envKey string, defaultValue int) int {
envVal := os.Getenv(envKey)
if envVal == "" {
return defaultValue
}
for pair := range strings.SplitSeq(checksumsStr, ",") {
parts := strings.SplitN(strings.TrimSpace(pair), ":", 2)
if len(parts) == 2 {
platform := strings.TrimSpace(parts[0])
hash := strings.TrimSpace(parts[1])
if platform != "" && hash != "" {
checksums[platform] = hash
}
}
parsed, err := strconv.Atoi(envVal)
if err != nil {
slog.Warn("Invalid int, using default", "env_key", envKey, "env_value", envVal, "default", defaultValue)
return defaultValue
}
return checksums
return parsed
}
pkg/appview/db/annotations_test.go
+1 -1
View File
@@ -21,7 +21,7 @@ func TestAnnotations_Placeholder(t *testing.T) {
func setupAnnotationsTestDB(t *testing.T) *sql.DB {
t.Helper()
// Use file::memory: with cache=shared to ensure all connections share the same in-memory DB
db, err := InitDB("file::memory:?cache=shared", true)
db, err := InitDB("file::memory:?cache=shared")
if err != nil {
t.Fatalf("Failed to initialize test database: %v", err)
}
pkg/appview/db/device_store_test.go
+1 -1
View File
@@ -14,7 +14,7 @@ func setupTestDB(t *testing.T) *DeviceStore {
t.Helper()
// Use file::memory: with cache=shared to ensure all connections share the same in-memory DB
// This prevents race conditions where different connections see different databases
db, err := InitDB("file::memory:?cache=shared", true)
db, err := InitDB("file::memory:?cache=shared")
if err != nil {
t.Fatalf("Failed to initialize test database: %v", err)
}
pkg/appview/db/hold_store_test.go
+1 -1
View File
@@ -81,7 +81,7 @@ func TestNullString(t *testing.T) {
func setupHoldTestDB(t *testing.T) *sql.DB {
t.Helper()
// Use file::memory: with cache=shared to ensure all connections share the same in-memory DB
db, err := InitDB("file::memory:?cache=shared", true)
db, err := InitDB("file::memory:?cache=shared")
if err != nil {
t.Fatalf("Failed to initialize test database: %v", err)
}
pkg/appview/db/oauth_store_test.go
+3 -3
View File
@@ -12,7 +12,7 @@ import (
func TestInvalidateSessionsWithMismatchedScopes(t *testing.T) {
// Create in-memory test database
db, err := InitDB(":memory:", true)
db, err := InitDB(":memory:")
if err != nil {
t.Fatalf("Failed to init database: %v", err)
}
@@ -232,7 +232,7 @@ func TestScopesMatch(t *testing.T) {
func TestOAuthStoreSessionLifecycle(t *testing.T) {
// Basic test to ensure SaveSession, GetSession, DeleteSession work correctly
db, err := InitDB(":memory:", true)
db, err := InitDB(":memory:")
if err != nil {
t.Fatalf("Failed to init database: %v", err)
}
@@ -304,7 +304,7 @@ func TestOAuthStoreSessionLifecycle(t *testing.T) {
}
func TestCleanupOldSessions(t *testing.T) {
db, err := InitDB(":memory:", true)
db, err := InitDB(":memory:")
if err != nil {
t.Fatalf("Failed to init database: %v", err)
}
pkg/appview/db/queries_test.go
+9 -9
View File
@@ -7,7 +7,7 @@ import (
func TestGetRepositoryMetadata(t *testing.T) {
// Create in-memory test database
db, err := InitDB(":memory:", true)
db, err := InitDB(":memory:")
if err != nil {
t.Fatalf("Failed to init database: %v", err)
}
@@ -143,7 +143,7 @@ func TestGetRepositoryMetadata(t *testing.T) {
func TestInsertManifest(t *testing.T) {
// Create in-memory test database
db, err := InitDB(":memory:", true)
db, err := InitDB(":memory:")
if err != nil {
t.Fatalf("Failed to init database: %v", err)
}
@@ -320,7 +320,7 @@ func TestInsertManifest(t *testing.T) {
func TestUserManagement(t *testing.T) {
// Create in-memory test database
db, err := InitDB(":memory:", true)
db, err := InitDB(":memory:")
if err != nil {
t.Fatalf("Failed to init database: %v", err)
}
@@ -432,7 +432,7 @@ func TestUserManagement(t *testing.T) {
func TestManifestOperations(t *testing.T) {
// Create in-memory test database
db, err := InitDB(":memory:", true)
db, err := InitDB(":memory:")
if err != nil {
t.Fatalf("Failed to init database: %v", err)
}
@@ -609,7 +609,7 @@ func TestManifestOperations(t *testing.T) {
func TestIsManifestTagged(t *testing.T) {
// Create in-memory test database
db, err := InitDB(":memory:", true)
db, err := InitDB(":memory:")
if err != nil {
t.Fatalf("Failed to init database: %v", err)
}
@@ -675,7 +675,7 @@ func TestIsManifestTagged(t *testing.T) {
func TestTagOperations(t *testing.T) {
// Create in-memory test database
db, err := InitDB(":memory:", true)
db, err := InitDB(":memory:")
if err != nil {
t.Fatalf("Failed to init database: %v", err)
}
@@ -838,7 +838,7 @@ func TestTagOperations(t *testing.T) {
func TestGetTagsWithPlatforms(t *testing.T) {
// Create in-memory test database
db, err := InitDB(":memory:", true)
db, err := InitDB(":memory:")
if err != nil {
t.Fatalf("Failed to init database: %v", err)
}
@@ -980,7 +980,7 @@ func TestGetTagsWithPlatforms(t *testing.T) {
func TestUpdateUserHandle(t *testing.T) {
// Create in-memory test database
db, err := InitDB(":memory:", true)
db, err := InitDB(":memory:")
if err != nil {
t.Fatalf("Failed to init database: %v", err)
}
@@ -1201,7 +1201,7 @@ func TestParseTimestamp(t *testing.T) {
}
func TestDeleteUserData(t *testing.T) {
db, err := InitDB(":memory:", true)
db, err := InitDB(":memory:")
if err != nil {
t.Fatalf("Failed to init database: %v", err)
}
pkg/appview/db/readonly.go
+2 -6
View File
@@ -57,11 +57,7 @@ func init() {
// InitializeDatabase initializes the SQLite database and session store
// Returns: (read-write DB, read-only DB, session store)
func InitializeDatabase(uiEnabled bool, dbPath string, skipMigrations bool) (*sql.DB, *sql.DB, *SessionStore) {
if !uiEnabled {
return nil, nil, nil
}
func InitializeDatabase(dbPath string) (*sql.DB, *sql.DB, *SessionStore) {
// Ensure directory exists
dbDir := filepath.Dir(dbPath)
if err := os.MkdirAll(dbDir, 0700); err != nil {
@@ -70,7 +66,7 @@ func InitializeDatabase(uiEnabled bool, dbPath string, skipMigrations bool) (*sq
}
// Initialize read-write database (for writes and auth operations)
database, err := InitDB(dbPath, skipMigrations)
database, err := InitDB(dbPath)
if err != nil {
slog.Warn("Failed to initialize UI database", "error", err)
return nil, nil, nil
pkg/appview/db/readonly_test.go
+1 -1
View File
@@ -19,7 +19,7 @@ func TestAuthorizerBlocksSensitiveTables(t *testing.T) {
defer os.Unsetenv("ATCR_UI_DATABASE_PATH")
// Initialize database (creates schema)
database, err := InitDB(dbPath, true)
database, err := InitDB(dbPath)
if err != nil {
t.Fatalf("Failed to initialize database: %v", err)
}
pkg/appview/db/schema.go
+4 -6
View File
@@ -26,7 +26,7 @@ var migrationsFS embed.FS
var schemaSQL string
// InitDB initializes the SQLite database with the schema
func InitDB(path string, skipMigrations bool) (*sql.DB, error) {
func InitDB(path string) (*sql.DB, error) {
db, err := sql.Open("sqlite3", path)
if err != nil {
return nil, err
@@ -54,12 +54,10 @@ func InitDB(path string, skipMigrations bool) (*sql.DB, error) {
}
}
// Run migrations unless skipped
// Run migrations
// For fresh databases, migrations are recorded but not executed (schema.sql is already complete)
if !skipMigrations {
if err := runMigrations(db, !isExisting); err != nil {
return nil, err
}
if err := runMigrations(db, !isExisting); err != nil {
return nil, err
}
return db, nil
pkg/appview/db/session_store_test.go
+1 -1
View File
@@ -13,7 +13,7 @@ import (
func setupSessionTestDB(t *testing.T) *SessionStore {
t.Helper()
// Use file::memory: with cache=shared to ensure all connections share the same in-memory DB
db, err := InitDB("file::memory:?cache=shared", true)
db, err := InitDB("file::memory:?cache=shared")
if err != nil {
t.Fatalf("Failed to initialize test database: %v", err)
}
pkg/appview/db/tag_delete_test.go
+1 -1
View File
@@ -11,7 +11,7 @@ import (
// This simulates what Jetstream does: encode repo/tag to rkey, then decode and delete
func TestTagDeleteRoundTrip(t *testing.T) {
// Create in-memory test database
db, err := InitDB(":memory:", true)
db, err := InitDB(":memory:")
if err != nil {
t.Fatalf("Failed to init database: %v", err)
}
pkg/appview/handlers/api.go
+7 -37
View File
@@ -6,7 +6,6 @@ import (
"fmt"
"log/slog"
"net/http"
"strings"
"atcr.io/pkg/appview/db"
"atcr.io/pkg/appview/middleware"
@@ -246,48 +245,19 @@ type CredentialHelperVersionResponse struct {
}
// CredentialHelperVersionHandler returns the latest credential helper version info
// Note: Version info is fetched dynamically from TangledRepo's releases
type CredentialHelperVersionHandler struct {
Version string
TangledRepo string
Checksums map[string]string
}
// Supported platforms for download URLs
var credentialHelperPlatforms = []struct {
key string // API key (e.g., "linux_amd64")
os string // OS name in archive (e.g., "Linux")
arch string // Arch name in archive (e.g., "x86_64")
ext string // Archive extension (e.g., "tar.gz" or "zip")
}{
{"linux_amd64", "Linux", "x86_64", "tar.gz"},
{"linux_arm64", "Linux", "arm64", "tar.gz"},
{"darwin_amd64", "Darwin", "x86_64", "tar.gz"},
{"darwin_arm64", "Darwin", "arm64", "tar.gz"},
{"windows_amd64", "Windows", "x86_64", "zip"},
{"windows_arm64", "Windows", "arm64", "zip"},
}
func (h *CredentialHelperVersionHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// Check if version is configured
if h.Version == "" {
http.Error(w, "Credential helper version not configured", http.StatusServiceUnavailable)
return
}
// Build download URLs for all platforms
// URL format: {TangledRepo}/tags/{version}/download/docker-credential-atcr_{version_without_v}_{OS}_{Arch}.{ext}
downloadURLs := make(map[string]string)
versionWithoutV := strings.TrimPrefix(h.Version, "v")
for _, p := range credentialHelperPlatforms {
filename := fmt.Sprintf("docker-credential-atcr_%s_%s_%s.%s", versionWithoutV, p.os, p.arch, p.ext)
downloadURLs[p.key] = fmt.Sprintf("%s/tags/%s/download/%s", h.TangledRepo, h.Version, filename)
}
// This endpoint directs users to the Tangled repository for downloads
// Version info should be fetched from the repository's releases page
response := CredentialHelperVersionResponse{
Latest: h.Version,
DownloadURLs: downloadURLs,
Checksums: h.Checksums,
Latest: "",
DownloadURLs: map[string]string{"tangled_repo": h.TangledRepo},
Checksums: nil,
ReleaseNotes: "Visit the Tangled repository for the latest releases: " + h.TangledRepo,
}
render.SetContentType(render.ContentTypeJSON)
pkg/appview/handlers/device_test.go
+1 -1
View File
@@ -18,7 +18,7 @@ import (
// setupTestDB creates an in-memory SQLite database with full schema for testing
func setupTestDB(t *testing.T) *sql.DB {
database, err := db.InitDB(":memory:", true)
database, err := db.InitDB(":memory:")
if err != nil {
t.Fatalf("Failed to initialize test database: %v", err)
}
pkg/appview/middleware/auth_test.go
+2 -2
View File
@@ -26,7 +26,7 @@ func TestGetUser_NoContext(t *testing.T) {
// setupTestDB creates an in-memory SQLite database for testing
func setupTestDB(t *testing.T) *sql.DB {
database, err := db.InitDB(":memory:", true)
database, err := db.InitDB(":memory:")
require.NoError(t, err)
t.Cleanup(func() {
@@ -307,7 +307,7 @@ func TestOptionalAuth_InvalidSession(t *testing.T) {
func TestMiddleware_ConcurrentAccess(t *testing.T) {
// Use a shared in-memory database for concurrent access
// (SQLite's default :memory: creates separate DBs per connection)
database, err := db.InitDB("file::memory:?cache=shared", true)
database, err := db.InitDB("file::memory:?cache=shared")
require.NoError(t, err)
t.Cleanup(func() {
database.Close()
pkg/auth/hold_remote_test.go
+1 -1
View File
@@ -34,7 +34,7 @@ func TestNewRemoteHoldAuthorizer_TestMode(t *testing.T) {
// setupTestDB creates an in-memory database for testing
func setupTestDB(t *testing.T) *sql.DB {
testDB, err := db.InitDB(":memory:", true)
testDB, err := db.InitDB(":memory:")
if err != nil {
t.Fatalf("Failed to initialize test database: %v", err)
}
pkg/auth/token/handler_test.go
+1 -1
View File
@@ -52,7 +52,7 @@ func getSharedTestKey(t *testing.T) string {
// setupTestDeviceStore creates an in-memory SQLite database for testing
func setupTestDeviceStore(t *testing.T) (*db.DeviceStore, *sql.DB) {
testDB, err := db.InitDB(":memory:", true)
testDB, err := db.InitDB(":memory:")
if err != nil {
t.Fatalf("Failed to initialize test database: %v", err)
}
pkg/hold/config.go
+63
View File
@@ -24,6 +24,7 @@ import (
type Config struct {
Version string `yaml:"version"`
LogLevel string `yaml:"log_level"`
LogShipper LogShipperConfig `yaml:"log_shipper"`
Storage StorageConfig `yaml:"storage"`
Server ServerConfig `yaml:"server"`
Registration RegistrationConfig `yaml:"registration"`
@@ -31,6 +32,28 @@ type Config struct {
Admin AdminConfig `yaml:"admin"`
}
// LogShipperConfig defines remote log shipping settings
type LogShipperConfig struct {
// Backend selects the log shipping backend (from env: ATCR_LOG_SHIPPER_BACKEND)
// Valid values: "victoria", "opensearch", "loki", or empty to disable
Backend string `yaml:"backend"`
// URL is the remote log service endpoint (from env: ATCR_LOG_SHIPPER_URL)
URL string `yaml:"url"`
// BatchSize is the number of logs to batch before flushing (from env: ATCR_LOG_SHIPPER_BATCH_SIZE, default: 100)
BatchSize int `yaml:"batch_size"`
// FlushInterval is the max time between flushes (from env: ATCR_LOG_SHIPPER_FLUSH_INTERVAL, default: 5s)
FlushInterval time.Duration `yaml:"flush_interval"`
// Username for basic auth (from env: ATCR_LOG_SHIPPER_USERNAME, optional)
Username string `yaml:"username"`
// Password for basic auth (from env: ATCR_LOG_SHIPPER_PASSWORD, optional)
Password string `yaml:"password"`
}
// AdminConfig defines admin panel settings
type AdminConfig struct {
// Enabled controls whether the admin panel is accessible (from env: HOLD_ADMIN_ENABLED)
@@ -114,6 +137,14 @@ func LoadConfigFromEnv() (*Config, error) {
// Logging configuration
cfg.LogLevel = getEnvOrDefault("ATCR_LOG_LEVEL", "info")
// Log shipper configuration
cfg.LogShipper.Backend = os.Getenv("ATCR_LOG_SHIPPER_BACKEND")
cfg.LogShipper.URL = os.Getenv("ATCR_LOG_SHIPPER_URL")
cfg.LogShipper.BatchSize = getIntOrDefault("ATCR_LOG_SHIPPER_BATCH_SIZE", 100)
cfg.LogShipper.FlushInterval = getDurationOrDefault("ATCR_LOG_SHIPPER_FLUSH_INTERVAL", 5*time.Second)
cfg.LogShipper.Username = os.Getenv("ATCR_LOG_SHIPPER_USERNAME")
cfg.LogShipper.Password = os.Getenv("ATCR_LOG_SHIPPER_PASSWORD")
// Server configuration
cfg.Server.Addr = getEnvOrDefault("HOLD_SERVER_ADDR", ":8080")
cfg.Server.PublicURL = os.Getenv("HOLD_PUBLIC_URL")
@@ -217,6 +248,38 @@ func getEnvOrDefault(key, defaultValue string) string {
return defaultValue
}
// getIntOrDefault parses an int from environment variable or returns default
func getIntOrDefault(envKey string, defaultValue int) int {
envVal := os.Getenv(envKey)
if envVal == "" {
return defaultValue
}
var parsed int
if _, err := fmt.Sscanf(envVal, "%d", &parsed); err != nil {
slog.Warn("Invalid int, using default", "env_key", envKey, "env_value", envVal, "default", defaultValue)
return defaultValue
}
return parsed
}
// getDurationOrDefault parses a duration from environment variable or returns default
func getDurationOrDefault(envKey string, defaultValue time.Duration) time.Duration {
envVal := os.Getenv(envKey)
if envVal == "" {
return defaultValue
}
parsed, err := time.ParseDuration(envVal)
if err != nil {
slog.Warn("Invalid duration, using default", "env_key", envKey, "env_value", envVal, "default", defaultValue)
return defaultValue
}
return parsed
}
// RequestCrawl sends a crawl request to the ATProto relay for the given hostname.
// This makes the hold's PDS discoverable by the relay network.
func RequestCrawl(relayEndpoint, publicURL string) error {
pkg/logging/logger.go
+42 -2
View File
@@ -27,6 +27,9 @@ var (
debugEnabled atomic.Bool
revertTimer *time.Timer
revertMu sync.Mutex
// asyncHandler holds the global async handler for shutdown
asyncHandler *AsyncHandler
)
// InitLogger initializes the global slog default logger with the specified log level.
@@ -36,6 +39,20 @@ var (
//
// Also starts a signal handler for SIGUSR1 to toggle debug mode at runtime.
func InitLogger(level string) {
InitLoggerWithShipper(level, ShipperConfig{})
}
// InitLoggerWithShipper initializes the global slog default logger with the specified
// log level and optional remote log shipping.
// Valid levels: debug, info, warn, error (case-insensitive)
// If level is empty or invalid, defaults to INFO.
// Call this from main() at startup.
//
// If shipperCfg.Backend is non-empty, logs will be shipped to the configured
// remote service in addition to stdout.
//
// Also starts a signal handler for SIGUSR1 to toggle debug mode at runtime.
func InitLoggerWithShipper(level string, shipperCfg ShipperConfig) {
var logLevel slog.Level
switch strings.ToLower(strings.TrimSpace(level)) {
@@ -69,13 +86,36 @@ func InitLogger(level string) {
},
}
handler := slog.NewTextHandler(os.Stdout, opts)
slog.SetDefault(slog.New(handler))
// Create stdout handler
stdoutHandler := slog.NewTextHandler(os.Stdout, opts)
// Create shipper if configured
var shipper Shipper
if shipperCfg.Backend != "" {
var err error
shipper, err = NewShipper(shipperCfg)
if err != nil {
// Log error but continue without shipping
fmt.Fprintf(os.Stderr, "log shipper initialization failed: %v (continuing with stdout only)\n", err)
}
}
// Create async handler (wraps stdout + optional shipper)
asyncHandler = NewAsyncHandler(stdoutHandler, shipper, shipperCfg, opts)
slog.SetDefault(slog.New(asyncHandler))
// Start signal handler for dynamic debug toggle
go handleDebugSignal()
}
// Shutdown flushes any remaining logs and closes the log shipper.
// Call this during graceful shutdown to ensure all logs are delivered.
func Shutdown() {
if asyncHandler != nil {
asyncHandler.Shutdown()
}
}
func handleDebugSignal() {
sigChan := make(chan os.Signal, 1)
signal.Notify(sigChan, syscall.SIGUSR1)
pkg/logging/shipper.go
+308
View File
@@ -0,0 +1,308 @@
// Package logging provides centralized structured logging with optional remote log shipping.
package logging
import (
"context"
"fmt"
"log/slog"
"sync"
"time"
)
// Default configuration values
const (
DefaultBatchSize = 100
DefaultFlushInterval = 5 * time.Second
)
// Shipper defines the interface for log shipping backends.
// Implementations should be safe for concurrent use.
type Shipper interface {
// Ship sends a batch of log entries to the remote service.
// Returns an error if the batch could not be shipped.
Ship(ctx context.Context, entries []LogEntry) error
// Close cleanly shuts down the shipper, releasing any resources.
Close() error
}
// LogEntry represents a single log entry to be shipped.
type LogEntry struct {
Time time.Time
Level slog.Level
Message string
Source string
Attrs map[string]any
}
// ShipperConfig configures the log shipper.
type ShipperConfig struct {
// Backend selects the shipping backend: "victoria", "opensearch", "loki", etc.
// Empty string disables remote shipping (stdout only).
Backend string
// URL is the remote service endpoint URL.
URL string
// BatchSize is the number of logs to batch before flushing.
// Default: 100
BatchSize int
// FlushInterval is the maximum time between flushes.
// Default: 5s
FlushInterval time.Duration
// Service identifies the source service ("appview" or "hold").
// Added to all log entries.
Service string
// Username for basic auth (optional).
Username string
// Password for basic auth (optional).
Password string
}
// NewShipper creates a shipper for the configured backend.
// Returns nil if no backend is configured (remote shipping disabled).
func NewShipper(cfg ShipperConfig) (Shipper, error) {
switch cfg.Backend {
case "victoria":
return NewVictoriaShipper(cfg)
case "opensearch":
return nil, fmt.Errorf("opensearch backend not yet implemented")
case "loki":
return nil, fmt.Errorf("loki backend not yet implemented")
case "":
return nil, nil // No remote shipping
default:
return nil, fmt.Errorf("unknown log shipper backend: %s", cfg.Backend)
}
}
// asyncState holds the shared state for async log shipping.
// This is separate from AsyncHandler to allow WithAttrs/WithGroup
// to create new handlers that share the same batch and flush state.
type asyncState struct {
shipper Shipper
// Batching
batch []LogEntry
batchMu sync.Mutex
batchSize int
// Async flush
flushInterval time.Duration
flushCh chan struct{}
doneCh chan struct{}
wg sync.WaitGroup
}
// AsyncHandler is an slog.Handler that writes to stdout and optionally
// ships logs to a remote service asynchronously.
type AsyncHandler struct {
stdout slog.Handler
opts *slog.HandlerOptions
state *asyncState // Shared state for batching and flushing
}
// NewAsyncHandler creates a new AsyncHandler that wraps stdout logging
// and optionally ships logs to a remote service.
func NewAsyncHandler(stdout slog.Handler, shipper Shipper, cfg ShipperConfig, opts *slog.HandlerOptions) *AsyncHandler {
batchSize := cfg.BatchSize
if batchSize <= 0 {
batchSize = DefaultBatchSize
}
flushInterval := cfg.FlushInterval
if flushInterval <= 0 {
flushInterval = DefaultFlushInterval
}
state := &asyncState{
shipper: shipper,
batch: make([]LogEntry, 0, batchSize),
batchSize: batchSize,
flushInterval: flushInterval,
flushCh: make(chan struct{}, 1),
doneCh: make(chan struct{}),
}
h := &AsyncHandler{
stdout: stdout,
opts: opts,
state: state,
}
// Start background flusher if shipping is enabled
if shipper != nil {
state.wg.Add(1)
go h.runFlusher()
}
return h
}
// Enabled reports whether the handler handles records at the given level.
func (h *AsyncHandler) Enabled(ctx context.Context, level slog.Level) bool {
return h.stdout.Enabled(ctx, level)
}
// Handle handles the Record by writing to stdout and queuing for remote shipping.
func (h *AsyncHandler) Handle(ctx context.Context, r slog.Record) error {
// Always write to stdout
if err := h.stdout.Handle(ctx, r); err != nil {
return err
}
// Skip remote shipping if no shipper configured
if h.state.shipper == nil {
return nil
}
// Build log entry
entry := LogEntry{
Time: r.Time,
Level: r.Level,
Message: r.Message,
Attrs: make(map[string]any),
}
r.Attrs(func(a slog.Attr) bool {
if a.Key == slog.SourceKey {
if src, ok := a.Value.Any().(*slog.Source); ok {
entry.Source = shortenSource(src.File, src.Line)
}
} else {
entry.Attrs[a.Key] = resolveAttrValue(a.Value)
}
return true
})
// Add to batch
h.state.batchMu.Lock()
h.state.batch = append(h.state.batch, entry)
shouldFlush := len(h.state.batch) >= h.state.batchSize
h.state.batchMu.Unlock()
if shouldFlush {
h.triggerFlush()
}
return nil
}
// WithAttrs returns a new Handler with the given attributes added.
func (h *AsyncHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
return &AsyncHandler{
stdout: h.stdout.WithAttrs(attrs),
opts: h.opts,
state: h.state, // Share the same state
}
}
// WithGroup returns a new Handler with the given group name.
func (h *AsyncHandler) WithGroup(name string) slog.Handler {
return &AsyncHandler{
stdout: h.stdout.WithGroup(name),
opts: h.opts,
state: h.state, // Share the same state
}
}
// triggerFlush signals the flusher goroutine to flush immediately.
func (h *AsyncHandler) triggerFlush() {
select {
case h.state.flushCh <- struct{}{}:
default: // Flush already pending
}
}
// runFlusher runs in a goroutine and periodically flushes the batch.
func (h *AsyncHandler) runFlusher() {
defer h.state.wg.Done()
ticker := time.NewTicker(h.state.flushInterval)
defer ticker.Stop()
for {
select {
case <-ticker.C:
h.flush()
case <-h.state.flushCh:
h.flush()
case <-h.state.doneCh:
h.flush() // Final flush
return
}
}
}
// flush sends the current batch to the remote service.
func (h *AsyncHandler) flush() {
h.state.batchMu.Lock()
if len(h.state.batch) == 0 {
h.state.batchMu.Unlock()
return
}
// Take ownership of the batch
batch := h.state.batch
h.state.batch = make([]LogEntry, 0, h.state.batchSize)
h.state.batchMu.Unlock()
// Ship with a timeout context
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if err := h.state.shipper.Ship(ctx, batch); err != nil {
// Log to stderr (not through slog to avoid recursion)
fmt.Printf("log shipper error: %v (dropped %d entries)\n", err, len(batch))
}
}
// Shutdown flushes any remaining logs and closes the shipper.
// Call this during graceful shutdown.
func (h *AsyncHandler) Shutdown() {
if h.state.shipper == nil {
return
}
close(h.state.doneCh)
h.state.wg.Wait()
if err := h.state.shipper.Close(); err != nil {
fmt.Printf("log shipper close error: %v\n", err)
}
}
// resolveAttrValue converts slog.Value to a plain Go value for JSON encoding.
func resolveAttrValue(v slog.Value) any {
switch v.Kind() {
case slog.KindString:
return v.String()
case slog.KindInt64:
return v.Int64()
case slog.KindUint64:
return v.Uint64()
case slog.KindFloat64:
return v.Float64()
case slog.KindBool:
return v.Bool()
case slog.KindDuration:
return v.Duration().String()
case slog.KindTime:
return v.Time().Format(time.RFC3339Nano)
case slog.KindGroup:
attrs := v.Group()
m := make(map[string]any, len(attrs))
for _, a := range attrs {
m[a.Key] = resolveAttrValue(a.Value)
}
return m
case slog.KindAny:
return v.Any()
default:
return v.String()
}
}
pkg/logging/shipper_test.go
+386
View File
@@ -0,0 +1,386 @@
package logging
import (
"context"
"encoding/json"
"io"
"log/slog"
"net/http"
"net/http/httptest"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
)
func TestNewShipper(t *testing.T) {
tests := []struct {
name string
cfg ShipperConfig
wantErr bool
errMsg string
}{
{
name: "empty backend returns nil",
cfg: ShipperConfig{Backend: ""},
wantErr: false,
},
{
name: "victoria backend requires URL",
cfg: ShipperConfig{Backend: "victoria", URL: ""},
wantErr: true,
errMsg: "URL is required",
},
{
name: "victoria backend with URL succeeds",
cfg: ShipperConfig{Backend: "victoria", URL: "http://localhost:9428"},
wantErr: false,
},
{
name: "unknown backend returns error",
cfg: ShipperConfig{Backend: "unknown"},
wantErr: true,
errMsg: "unknown log shipper backend",
},
{
name: "opensearch not implemented",
cfg: ShipperConfig{Backend: "opensearch"},
wantErr: true,
errMsg: "not yet implemented",
},
{
name: "loki not implemented",
cfg: ShipperConfig{Backend: "loki"},
wantErr: true,
errMsg: "not yet implemented",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
shipper, err := NewShipper(tt.cfg)
if tt.wantErr {
if err == nil {
t.Errorf("NewShipper() expected error, got nil")
} else if tt.errMsg != "" && !strings.Contains(err.Error(), tt.errMsg) {
t.Errorf("NewShipper() error = %v, want error containing %q", err, tt.errMsg)
}
return
}
if err != nil {
t.Errorf("NewShipper() unexpected error: %v", err)
}
if tt.cfg.Backend == "" && shipper != nil {
t.Error("NewShipper() with empty backend should return nil shipper")
}
if shipper != nil {
shipper.Close()
}
})
}
}
func TestVictoriaShipper_Ship(t *testing.T) {
var receivedLogs []map[string]any
var mu sync.Mutex
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
t.Errorf("expected POST, got %s", r.Method)
}
if !strings.Contains(r.URL.Path, "/insert/jsonline") {
t.Errorf("expected /insert/jsonline path, got %s", r.URL.Path)
}
if r.Header.Get("Content-Type") != "application/stream+json" {
t.Errorf("expected application/stream+json content type, got %s", r.Header.Get("Content-Type"))
}
body, _ := io.ReadAll(r.Body)
lines := strings.Split(strings.TrimSpace(string(body)), "\n")
mu.Lock()
for _, line := range lines {
var doc map[string]any
if err := json.Unmarshal([]byte(line), &doc); err != nil {
t.Errorf("failed to unmarshal log line: %v", err)
}
receivedLogs = append(receivedLogs, doc)
}
mu.Unlock()
w.WriteHeader(http.StatusOK)
}))
defer server.Close()
shipper, err := NewVictoriaShipper(ShipperConfig{
URL: server.URL,
Service: "test-service",
})
if err != nil {
t.Fatalf("NewVictoriaShipper() error: %v", err)
}
defer shipper.Close()
entries := []LogEntry{
{
Time: time.Date(2024, 1, 8, 12, 0, 0, 0, time.UTC),
Level: slog.LevelInfo,
Message: "test message 1",
Source: "test.go:42",
Attrs: map[string]any{"key1": "value1"},
},
{
Time: time.Date(2024, 1, 8, 12, 0, 1, 0, time.UTC),
Level: slog.LevelError,
Message: "test message 2",
Source: "test.go:43",
Attrs: map[string]any{"key2": 123},
},
}
ctx := context.Background()
if err := shipper.Ship(ctx, entries); err != nil {
t.Fatalf("Ship() error: %v", err)
}
mu.Lock()
defer mu.Unlock()
if len(receivedLogs) != 2 {
t.Errorf("expected 2 logs, got %d", len(receivedLogs))
}
// Check first log
if receivedLogs[0]["_msg"] != "test message 1" {
t.Errorf("expected _msg 'test message 1', got %v", receivedLogs[0]["_msg"])
}
if receivedLogs[0]["level"] != "INFO" {
t.Errorf("expected level 'INFO', got %v", receivedLogs[0]["level"])
}
if receivedLogs[0]["service"] != "test-service" {
t.Errorf("expected service 'test-service', got %v", receivedLogs[0]["service"])
}
if receivedLogs[0]["key1"] != "value1" {
t.Errorf("expected key1 'value1', got %v", receivedLogs[0]["key1"])
}
// Check second log
if receivedLogs[1]["level"] != "ERROR" {
t.Errorf("expected level 'ERROR', got %v", receivedLogs[1]["level"])
}
}
func TestVictoriaShipper_BasicAuth(t *testing.T) {
var authHeader string
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
authHeader = r.Header.Get("Authorization")
w.WriteHeader(http.StatusOK)
}))
defer server.Close()
shipper, err := NewVictoriaShipper(ShipperConfig{
URL: server.URL,
Username: "testuser",
Password: "testpass",
})
if err != nil {
t.Fatalf("NewVictoriaShipper() error: %v", err)
}
defer shipper.Close()
entries := []LogEntry{{Time: time.Now(), Level: slog.LevelInfo, Message: "test"}}
if err := shipper.Ship(context.Background(), entries); err != nil {
t.Fatalf("Ship() error: %v", err)
}
if authHeader == "" {
t.Error("expected Authorization header to be set")
}
if !strings.HasPrefix(authHeader, "Basic ") {
t.Errorf("expected Basic auth, got: %s", authHeader)
}
}
func TestVictoriaShipper_EmptyBatch(t *testing.T) {
var called bool
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
called = true
w.WriteHeader(http.StatusOK)
}))
defer server.Close()
shipper, _ := NewVictoriaShipper(ShipperConfig{URL: server.URL})
defer shipper.Close()
if err := shipper.Ship(context.Background(), nil); err != nil {
t.Errorf("Ship() with nil entries should not error: %v", err)
}
if err := shipper.Ship(context.Background(), []LogEntry{}); err != nil {
t.Errorf("Ship() with empty entries should not error: %v", err)
}
if called {
t.Error("Ship() should not make HTTP request for empty batch")
}
}
func TestVictoriaShipper_ServerError(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte("internal error"))
}))
defer server.Close()
shipper, _ := NewVictoriaShipper(ShipperConfig{URL: server.URL})
defer shipper.Close()
entries := []LogEntry{{Time: time.Now(), Level: slog.LevelInfo, Message: "test"}}
err := shipper.Ship(context.Background(), entries)
if err == nil {
t.Error("Ship() should return error on server error")
}
if !strings.Contains(err.Error(), "500") {
t.Errorf("error should contain status code, got: %v", err)
}
}
func TestAsyncHandler_Batching(t *testing.T) {
var shipCount atomic.Int32
var totalEntries atomic.Int32
mockShipper := &mockShipper{
shipFunc: func(ctx context.Context, entries []LogEntry) error {
shipCount.Add(1)
totalEntries.Add(int32(len(entries)))
return nil
},
}
cfg := ShipperConfig{
BatchSize: 5,
FlushInterval: 100 * time.Millisecond,
}
stdoutHandler := slog.NewTextHandler(io.Discard, nil)
handler := NewAsyncHandler(stdoutHandler, mockShipper, cfg, nil)
// Log 12 entries - should trigger 2 batch flushes (at 5 and 10) plus 2 remaining
for i := 0; i < 12; i++ {
record := slog.NewRecord(time.Now(), slog.LevelInfo, "test message", 0)
handler.Handle(context.Background(), record)
}
// Wait for flush interval to trigger
time.Sleep(200 * time.Millisecond)
handler.Shutdown()
if totalEntries.Load() != 12 {
t.Errorf("expected 12 total entries shipped, got %d", totalEntries.Load())
}
}
func TestAsyncHandler_Shutdown(t *testing.T) {
var shipped []LogEntry
var mu sync.Mutex
mockShipper := &mockShipper{
shipFunc: func(ctx context.Context, entries []LogEntry) error {
mu.Lock()
shipped = append(shipped, entries...)
mu.Unlock()
return nil
},
}
cfg := ShipperConfig{
BatchSize: 100, // Large batch size so nothing flushes immediately
FlushInterval: 10 * time.Second,
}
stdoutHandler := slog.NewTextHandler(io.Discard, nil)
handler := NewAsyncHandler(stdoutHandler, mockShipper, cfg, nil)
// Log a few entries
for i := 0; i < 3; i++ {
record := slog.NewRecord(time.Now(), slog.LevelInfo, "test", 0)
handler.Handle(context.Background(), record)
}
// Entries should be pending (not shipped yet due to large batch size)
mu.Lock()
if len(shipped) != 0 {
t.Errorf("expected 0 shipped before shutdown, got %d", len(shipped))
}
mu.Unlock()
// Shutdown should flush pending entries
handler.Shutdown()
mu.Lock()
if len(shipped) != 3 {
t.Errorf("expected 3 shipped after shutdown, got %d", len(shipped))
}
mu.Unlock()
}
func TestAsyncHandler_NoShipper(t *testing.T) {
cfg := ShipperConfig{}
stdoutHandler := slog.NewTextHandler(io.Discard, nil)
handler := NewAsyncHandler(stdoutHandler, nil, cfg, nil)
// Should not panic with nil shipper
record := slog.NewRecord(time.Now(), slog.LevelInfo, "test", 0)
if err := handler.Handle(context.Background(), record); err != nil {
t.Errorf("Handle() with nil shipper should not error: %v", err)
}
// Shutdown should not panic
handler.Shutdown()
}
func TestResolveAttrValue(t *testing.T) {
tests := []struct {
name string
value slog.Value
expected any
}{
{"string", slog.StringValue("test"), "test"},
{"int64", slog.Int64Value(42), int64(42)},
{"uint64", slog.Uint64Value(42), uint64(42)},
{"float64", slog.Float64Value(3.14), 3.14},
{"bool", slog.BoolValue(true), true},
{"duration", slog.DurationValue(5 * time.Second), "5s"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := resolveAttrValue(tt.value)
if got != tt.expected {
t.Errorf("resolveAttrValue() = %v, want %v", got, tt.expected)
}
})
}
}
// mockShipper implements Shipper for testing
type mockShipper struct {
shipFunc func(ctx context.Context, entries []LogEntry) error
closeFunc func() error
}
func (m *mockShipper) Ship(ctx context.Context, entries []LogEntry) error {
if m.shipFunc != nil {
return m.shipFunc(ctx, entries)
}
return nil
}
func (m *mockShipper) Close() error {
if m.closeFunc != nil {
return m.closeFunc()
}
return nil
}
pkg/logging/victoria.go
+112
View File
@@ -0,0 +1,112 @@
package logging
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"time"
)
// VictoriaShipper ships logs to Victoria Logs using the native JSON lines endpoint.
type VictoriaShipper struct {
url string
client *http.Client
service string
username string
password string
}
// NewVictoriaShipper creates a new Victoria Logs shipper.
func NewVictoriaShipper(cfg ShipperConfig) (*VictoriaShipper, error) {
if cfg.URL == "" {
return nil, fmt.Errorf("victoria logs URL is required")
}
return &VictoriaShipper{
url: cfg.URL,
service: cfg.Service,
client: &http.Client{
Timeout: 30 * time.Second,
},
username: cfg.Username,
password: cfg.Password,
}, nil
}
// Ship sends a batch of log entries to Victoria Logs.
func (v *VictoriaShipper) Ship(ctx context.Context, entries []LogEntry) error {
if len(entries) == 0 {
return nil
}
var buf bytes.Buffer
for _, entry := range entries {
doc := map[string]any{
// Victoria Logs special fields
"_time": entry.Time.UTC().Format(time.RFC3339Nano),
"_msg": entry.Message,
// Standard fields
"level": entry.Level.String(),
"source": entry.Source,
}
// Add service if configured
if v.service != "" {
doc["service"] = v.service
}
// Add all custom attributes
for k, val := range entry.Attrs {
// Don't overwrite special fields
if k != "_time" && k != "_msg" && k != "level" && k != "source" && k != "service" {
doc[k] = val
}
}
if err := json.NewEncoder(&buf).Encode(doc); err != nil {
return fmt.Errorf("failed to encode log entry: %w", err)
}
}
// Use the JSON lines endpoint with stream fields for efficient querying
url := v.url + "/insert/jsonline?_stream_fields=service,level"
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, &buf)
if err != nil {
return fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/stream+json")
// Add basic auth if configured
if v.username != "" && v.password != "" {
req.SetBasicAuth(v.username, v.password)
}
resp, err := v.client.Do(req)
if err != nil {
return fmt.Errorf("failed to send logs: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode >= 400 {
body, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
return fmt.Errorf("victoria logs error %d: %s", resp.StatusCode, string(body))
}
// Drain and close body to allow connection reuse
_, _ = io.Copy(io.Discard, resp.Body)
return nil
}
// Close releases any resources held by the shipper.
func (v *VictoriaShipper) Close() error {
v.client.CloseIdleConnections()
return nil
}