diff --git a/pkg/mongo/mongo.go b/pkg/mongo/mongo.go index 0d0af80..0126eb9 100644 --- a/pkg/mongo/mongo.go +++ b/pkg/mongo/mongo.go @@ -17,7 +17,18 @@ func GlobalTimeout(ctx context.Context) context.Context { return ctx } -func Initialize(ctx context.Context, uri string, opts *mongoOpt.ClientOptions) (*mongo.Collection, error) { +type InitializeOptions struct { + MaxPoolSize uint64 + CapSizeBytes int64 + ExpireAfterSeconds int64 +} + +func Initialize(ctx context.Context, uri string, opt *InitializeOptions) (*mongo.Collection, error) { + collectionName := "logs" + if opt == nil { + opt = &InitializeOptions{} + } + uriParsed, err := url.ParseRequestURI(uri) if err != nil { return nil, fmt.Errorf("parsing URI for database name: %w", err) @@ -28,7 +39,7 @@ func Initialize(ctx context.Context, uri string, opts *mongoOpt.ClientOptions) ( return nil, fmt.Errorf("URI must include database name (as database to authenticate against)") } - dbOpt := attachMetrics(opts).ApplyURI(uri) + dbOpt := attachMetrics(mongoOpt.Client()).ApplyURI(uri).SetMaxPoolSize(opt.MaxPoolSize) dbClient, err := mongo.Connect(GlobalTimeout(ctx), dbOpt) if err != nil { @@ -39,7 +50,19 @@ func Initialize(ctx context.Context, uri string, opts *mongoOpt.ClientOptions) ( return nil, fmt.Errorf("first ping to database: %w", err) } - col := dbClient.Database(uriParsed.Path).Collection("logs") + db := dbClient.Database(uriParsed.Path) + + capped := opt.CapSizeBytes > 0 + + if err := db.CreateCollection(GlobalTimeout(ctx), collectionName, &mongoOpt.CreateCollectionOptions{ + Capped: &capped, + SizeInBytes: &opt.CapSizeBytes, + ExpireAfterSeconds: &opt.ExpireAfterSeconds, + }); err != nil { + return nil, fmt.Errorf("initializing collection") + } + + col := db.Collection(collectionName) if err := InitializeIndexes(GlobalTimeout(ctx), col); err != nil { return nil, fmt.Errorf("initializing indexes: %w", err) diff --git a/pkg/watcher/app.go b/pkg/watcher/app.go index 67d99ab..5947eda 100644 --- a/pkg/watcher/app.go +++ b/pkg/watcher/app.go @@ -14,7 +14,6 @@ import ( "git.k-space.ee/k-space/logmower-shipper/pkg/util" "github.com/fsnotify/fsnotify" "github.com/urfave/cli/v2" - mongoOpt "go.mongodb.org/mongo-driver/mongo/options" ) var App = &cli.App{ @@ -27,20 +26,22 @@ var App = &cli.App{ Flags: []cli.Flag{ // in Action &cli.StringFlag{Name: "log-directory", Usage: "Directory to watch for logs", Value: "/var/log/containers"}, - &cli.Uint64Flag{Name: "max-connection-pool-size", EnvVars: []string{"MAX_CONNECTION_POOL_SIZE"}, Value: 1, Usage: "Max MongoDB connection pool size"}, + &cli.Uint64Flag{Category: "mongo", Name: "max-connection-pool-size", EnvVars: []string{"MAX_CONNECTION_POOL_SIZE"}, Value: 1, Usage: "Max MongoDB connection pool size"}, + &cli.Int64Flag{Category: "mongo", Name: "max-collection-size", EnvVars: []string{"MAX_COLLECTION_SIZE"}, Usage: "MongoDB collection size limit in bytes"}, + &cli.Int64Flag{Category: "mongo", Name: "max-record-retention", EnvVars: []string{"MAX_RECORD_RETENTION"}, Usage: "Record retention in seconds"}, + &cli.StringFlag{Category: "mongo", Name: "mongo-uri", EnvVars: []string{"MONGODB_URI"}, Usage: "mongodb://foo:bar@host:27017/database", Required: true}, // in Before - &cli.BoolFlag{Name: "simulate", Aliases: []string{"dry-run"}, Usage: "Do not write to database"}, + &cli.BoolFlag{Category: "mongo", Name: "simulate", Aliases: []string{"dry-run"}, Usage: "Writes to database are simulate as successful"}, &cli.IntFlag{Name: "max-record-size", EnvVars: []string{"MAX_RECORD_SIZE"}, Value: 128 * 1024, Usage: "Maximum record size in bytes"}, &cli.IntFlag{Name: "bulk-insertion-size", EnvVars: []string{"BULK_INSERTION_SIZE"}, Value: 1000, Usage: "MongoDB bulk insertion size in records"}, &cli.IntFlag{Name: "max-upload-queue-size", EnvVars: []string{"MAX_UPLOAD_QUEUE_SIZE"}, Value: 1024, Usage: "Max upload queue size (before batching) in records"}, - // - //TODO: &cli.BoolFlag{Name: "heuristic-normalize-log-level", Usage: "Normalize log.level values to Syslog defined keywords", Value: false}, - //TODO: &cli.BoolFlag{Name: "heuristic-parse-json", Usage: "Attempt automatically unwrapping JSON records", Value: false}, + + // TODO: &cli.BoolFlag{Category: "parsing", Name: "heuristic-normalize-log-level", Usage: "Normalize log.level values to Syslog defined keywords", Value: false}, + // TODO: &cli.BoolFlag{Category: "parsing", Name: "heuristic-parse-json", Usage: "Attempt automatically unwrapping JSON records", Value: false}, &cli.StringSliceFlag{Category: "selectors", Name: "namespace", EnvVars: []string{"NAMESPACE"}, Usage: "whitelist filter for filenames"}, &cli.StringSliceFlag{Category: "selectors", Name: "exclude-pod-prefixes", EnvVars: []string{"EXCLUDE_POD_PREFIXES"}, Usage: "blacklist filter for filenames", Value: cli.NewStringSlice("logmower-")}, // - &cli.StringFlag{Category: "secrets", Name: "mongo-uri", EnvVars: []string{"MONGODB_URI"}, Usage: "mongodb://foo:bar@host:27017/database", Required: true}, }, Before: func(ctx *cli.Context) error { sender.Simulate = ctx.Bool("simulate") @@ -60,8 +61,11 @@ var App = &cli.App{ log.Printf("%s %s starting", ctx.App.Name, ctx.App.Version) - db, err := m.Initialize(ctx.Context, ctx.String("mongo-uri"), mongoOpt.Client(). - SetMaxPoolSize(ctx.Uint64("max-connection-pool-size"))) + db, err := m.Initialize(ctx.Context, ctx.String("mongo-uri"), &m.InitializeOptions{ + MaxPoolSize: ctx.Uint64("max-connection-pool-size"), + CapSizeBytes: ctx.Int64("max-collection-size"), + ExpireAfterSeconds: ctx.Int64("max-record-retention"), + }) if err != nil { return fmt.Errorf("initializing database connection: %w", err) } @@ -95,7 +99,6 @@ var App = &cli.App{ continue } - // TODO: #1: || if not in filterset kubeInfo, ok := util.ParseLogFilename(event.Name) if !ok { promWatcherFilesSkipped.Add(1)