150 lines
3.1 KiB
Go
150 lines
3.1 KiB
Go
package mongo
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"go.mongodb.org/mongo-driver/bson"
|
|
"go.mongodb.org/mongo-driver/mongo"
|
|
)
|
|
|
|
// ctx is used directly
|
|
func InitializeIndexes(ctx context.Context, col *mongo.Collection) error {
|
|
ind := col.Indexes()
|
|
|
|
// (does not create duplicates)
|
|
_, err := ind.CreateOne(ctx, mongo.IndexModel{
|
|
Keys: bson.D{{Key: RecordKeyFilePath, Value: 1}, {Key: RecordKeyOffset, Value: -1}},
|
|
})
|
|
|
|
return err
|
|
}
|
|
|
|
// when editing, also edit everything in this file!
|
|
type (
|
|
Record struct {
|
|
*File
|
|
Offset int64 // end, of last line
|
|
|
|
String string
|
|
ParsedMetadata
|
|
|
|
ParsedContent // TODO: ECS
|
|
|
|
// added by ToBson()
|
|
timeShip time.Time
|
|
}
|
|
ParsedMetadata struct {
|
|
TimeKubernetes time.Time
|
|
StdErr bool
|
|
}
|
|
ParsedContent struct {
|
|
Content any
|
|
TimeUpstream time.Time
|
|
}
|
|
Host struct {
|
|
Id string
|
|
Name string
|
|
Arch string
|
|
}
|
|
File struct {
|
|
Host *Host
|
|
Path string // absolute
|
|
KubeInfo
|
|
}
|
|
KubeInfo struct {
|
|
ContainerName string
|
|
ContainerId string // unused
|
|
Namespace string
|
|
Pod string
|
|
}
|
|
)
|
|
|
|
const (
|
|
// used outside package for mongo commands
|
|
RecordKeyHostId = recordKeyHost + "." + recordKeyId
|
|
RecordKeyFilePath = recordKeyLog + "." + recordKeyFile + "." + recordKeygenericPath
|
|
RecordKeyOffset = recordKeyLog + "." + recordKeyOffset
|
|
)
|
|
|
|
// Don't use direct strings in bson types. Use the constants as keys.
|
|
// This ensures keys (and subkeys) are consistent within the package, and by consumers of it.
|
|
|
|
const (
|
|
recordKeygenericName = "name"
|
|
recordKeygenericPath = "path"
|
|
)
|
|
|
|
const (
|
|
recordKeyString = "message"
|
|
|
|
recordKeyLog = "log"
|
|
recordKeyFile = "file"
|
|
recordKeyOffset = "offset"
|
|
// recordKeyLevel = "level"
|
|
|
|
recordKeyHost = "host"
|
|
recordKeyId = "id"
|
|
recordKeyName = "name"
|
|
recordKeyArch = "architecture"
|
|
|
|
recordKeyKubernetes = "kubernetes"
|
|
recordKeyContainer = "container"
|
|
recordKeyNamespace = "namespace"
|
|
recordKeyPod = "pod"
|
|
|
|
recordKeyStream = "stream"
|
|
|
|
recordKeyEvent = "event"
|
|
recordKeyTimeUpstream = "created"
|
|
recordKeyTimeKubernetes = "ingested"
|
|
|
|
recordKeyTimeMower = "@timestamp"
|
|
)
|
|
|
|
// not using marshal, since it is <0.1x performance
|
|
func (l *Record) ToBson() bson.M {
|
|
var stream string
|
|
if l.StdErr {
|
|
stream = "stderr"
|
|
} else {
|
|
stream = "stdout"
|
|
}
|
|
|
|
return bson.M{
|
|
recordKeyString: l.String,
|
|
recordKeyLog: bson.M{
|
|
recordKeyFile: bson.M{
|
|
recordKeygenericPath: l.File.Path,
|
|
},
|
|
recordKeyOffset: l.Offset,
|
|
// recordKeyLevel: , //TODO: ECS
|
|
},
|
|
recordKeyKubernetes: bson.M{
|
|
recordKeyContainer: bson.M{
|
|
recordKeygenericName: l.File.ContainerName,
|
|
},
|
|
recordKeyNamespace: l.File.Namespace,
|
|
recordKeyPod: bson.M{
|
|
recordKeygenericName: l.File.Pod,
|
|
},
|
|
},
|
|
recordKeyHost: bson.M{
|
|
recordKeyId: l.File.Host.Id,
|
|
recordKeyName: l.File.Host.Name,
|
|
recordKeyArch: l.File.Host.Arch,
|
|
},
|
|
recordKeyStream: stream,
|
|
recordKeyEvent: bson.M{
|
|
// recordKeyTimeUpstream: l.TimeUpstream, //TODO: ECS
|
|
recordKeyTimeKubernetes: l.TimeKubernetes,
|
|
},
|
|
|
|
recordKeyTimeMower: time.Now(),
|
|
}
|
|
}
|
|
|
|
func RecordOffsetFromBson(b *bson.Raw) int64 {
|
|
return bsonLookupInt64(b, recordKeyLog, recordKeyOffset)
|
|
}
|