ingest

package
v0.0.0-...-e5e9e03 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 20, 2026 License: Apache-2.0 Imports: 51 Imported by: 0

README

LogIngest

Flow

init->download->unpack->preprocess->dbconnect->processlogs->processcf
init->(progress-tracker)

Components

name description
init initialize a new ingest system
download download files from s3/sftp/aerolab cluster
unpack (+compression) unpack the files recursively
preprocess identify files as logs/collectinfo/other, rename and store collectinfo, rename and store logs according to discovered cluster name and node ID; also handle custom log formats (json, etc)
preprocess-special special (tab/json) format handler
dbconnect connect to the backend database
processlogs actually process the logs, inserting the statistics into the embedded agi/db store
processcf find out which node the collectinfo was gathered on, and rename the files accordingly to match log prefixes
trackprogress track progress in a json file and to stderr/out
enum enumerate through the dirty directory; used by multiple components
logstream handle logs as singular streams (one per log file) by actually extracting patterns and timestamp and returning the result

Documentation

Index

Constants

View Source 
const MetricsRowPKSeparator = "::/::"

MetricsRowPKSeparator is the legacy three-part-key delimiter used inside the hash input. It is intentionally identical to the unhashed v2 separator so re-ingest of the same source data produces the same hash and therefore the same idempotent overwrite semantics ingest has always relied on.

Variables

This section is empty.

Functions

func ClearDirtyMarker 

func ClearDirtyMarker(progressDir string) error

ClearDirtyMarker removes the sentinel. It is a no-op when the marker is already absent (clean re-runs / fresh hosts) and reports any other os error to the caller. Callers that want fire-and-forget semantics can ignore the return value — leaving a stale marker behind only causes one extra wipe-and-reingest on the next startup.

func DBOptionsFromConfig 

func DBOptionsFromConfig(cfg *Config) db.Options

DBOptionsFromConfig translates the ingest Config's DB sub-struct into a db.Options. Callers that share a DB handle across packages (see cmdAgiExecService) use this to build the same options without duplicating field-by-field mapping. Mirrors plugin.DBOptionsFromConfig: numeric zero is the "not set" sentinel (preserve db.DefaultOptions); bool fields are operator-authoritative because Go bools cannot represent "not set".

func DirtyMarkerExists 

func DirtyMarkerExists(progressDir string) bool

DirtyMarkerExists reports whether the sentinel is present at DirtyMarkerPath(progressDir). Any stat error other than os.IsNotExist is treated conservatively as "exists" so a transient EFS glitch never silently downgrades a real dirty state to "clean".

func DirtyMarkerPath 

func DirtyMarkerPath(progressDir string) string

DirtyMarkerPath returns the on-disk path of the in-progress sentinel for the given progress directory. The path is the same regardless of whether the directory exists yet — callers that want to write the marker should MkdirAll on the parent first.

func MetricsRowKey 

func MetricsRowKey(clusterName, nodeIdent, logLine string) string

MetricsRowKey returns the deterministic XXH3-128 hash of the (cluster, node, log-line) tuple used as the metrics-set primary key. The components are joined by MetricsRowPKSeparator before hashing.

XXH3-128 is non-cryptographic but provides a 128-bit output; the birthday-bound collision probability for 10^10 distinct keys (~1 TiB of typical Aerospike server logs at ~150 B/line) is ≈1.5e-19, several orders of magnitude below the host's other failure modes (uncorrected disk read errors, ECC RAM faults, etc.). 64-bit hashes are *not* sufficient at this scale; 128 bits gives roughly 2^64 birthday-safety headroom.

Output is 32 lowercase hex characters (16 raw bytes encoded). We encode rather than using the raw bytes as the DB key so the key remains a printable Go string and so debug tools (logs, /debug/db/sample, --get-key) can echo it without escaping.

func MetricsRowKeyFromCombined 

func MetricsRowKeyFromCombined(nodeIdentifier, logLine string) string

MetricsRowKeyFromCombined is the same hash computed from a pre-joined "<cluster><sep><node>" identifier and a log line. Equivalent to MetricsRowKey when the join uses MetricsRowPKSeparator; ingest's hot path takes this form because the combined string is already constructed once per log file.

func MetricsRowKeyFromString 

func MetricsRowKeyFromString(joined string) string

MetricsRowKeyFromString hashes an arbitrary pre-joined string in legacy form ("cluster::/::node::/::line"). Exposed for the `aerolab agi query --hash-key` debugging helper so operators can compute the same key ingest would have produced for a given combined input.

func Run 

func Run(yamlFile string) error

Run executes the complete log ingestion pipeline using configuration from a YAML file. This is the main entry point for the log ingestion system that handles the entire workflow from downloading logs to processing and storing them in the embedded agi/db store.

The function performs the following steps: 1. Load configuration from YAML file and environment variables 2. Initialize the ingestion system with database connections 3. Download logs from configured sources (S3, SFTP, local) 4. Unpack and decompress log files 5. Preprocess logs to identify clusters and nodes 6. Process logs and collectinfo files concurrently 7. Store processed data in the embedded agi/db store

Parameters:

  • yamlFile: Path to the YAML configuration file. If empty, uses environment variables only.

Returns:

  • error: nil on success, or an error describing what failed during ingestion

Usage: 

err := ingest.Run("config.yaml")
if err != nil {
    log.Fatal("Ingestion failed:", err)
}

func RunWithConfig 

func RunWithConfig(config *Config) error

RunWithConfig executes the complete log ingestion pipeline using a pre-configured Config object. This function provides more control than Run() by allowing direct configuration without file parsing. It performs the same ingestion workflow but uses the provided configuration directly.

The ingestion process runs log processing and collectinfo processing concurrently for better performance. If either process fails, the function will return an error with details about all failures.

Parameters:

  • config: Pre-configured ingestion configuration with all necessary settings

Returns:

  • error: nil on success, or a combined error if any step fails

Usage: 

config := &ingest.Config{
    DB: dbConfig,
    Downloader: downloaderConfig,
    // ... other settings
}
err := ingest.RunWithConfig(config)
if err != nil {
    log.Fatal("Ingestion failed:", err)
}

func SftpCheckLogin 

func SftpCheckLogin(config *Config, getFileList bool) (map[string]*DownloaderFile, error)

func WriteDirtyMarker 

func WriteDirtyMarker(progressDir string) error

WriteDirtyMarker creates (or refreshes) the sentinel file at DirtyMarkerPath(progressDir). It is safe to call multiple times per run; the second call is a no-op when the file is already present.

Failure is non-fatal in spirit (the worst case is that a future crash-on-WAL-off run misses the wipe and re-ingests one extra time), but we surface the error so the operator sees it. Callers that prefer best-effort can ignore the return value.

Types

type CfFile 

type CfFile struct {
	Size                int64
	NodeID              string
	RenameAttempted     bool
	Renamed             bool
	OriginalName        string
	ProcessingAttempted bool
	Processed           bool
	Errors              []string
	StartTime           string
	FinishTime          string
}

type Config 

type Config struct {
	LogLevel int `yaml:"logLevel" default:"4" envconfig:"LOGINGEST_LOGLEVEL"` // 0=NO_LOGGING 1=CRITICAL, 2=ERROR, 3=WARNING, 4=INFO, 5=DEBUG, 6=DETAIL
	// DB holds embedded-db tuning knobs only. Pipeline knobs that used
	// to live here (DefaultSetName, LogFileRangesSetName,
	// TimestampColumnName, MaxPutThreads) were moved to top-level
	// fields below — they are not properties of the storage engine.
	DB struct {
		// Path is the on-disk directory Pebble writes to. The default
		// is db.DefaultPath; keep ingest and plugin in lockstep or
		// they will open separate stores and never see each other's
		// data.
		Path                        string `yaml:"path" default:"/opt/agi/db" envconfig:"LOGINGEST_DB_PATH"`
		CacheBytes                  int64  `yaml:"cacheBytes" default:"0"`                  // 0 -> db default
		MemTableSizeBytes           uint64 `yaml:"memTableSizeBytes" default:"0"`           // 0 -> db default
		MemTableStopWritesThreshold int    `yaml:"memTableStopWritesThreshold" default:"0"` // 0 -> db default
		MaxConcurrentCompactions    int    `yaml:"maxConcurrentCompactions" default:"0"`    // 0 -> db default
		MaxOpenFiles                int    `yaml:"maxOpenFiles" default:"0"`
		BlockSize                   int    `yaml:"blockSize" default:"0"`  // 0 -> db default (Pebble default = 4 KiB)
		Compression                 string `yaml:"compression" default:""` // "" -> db default (Pebble default = uniform Snappy); see db.Options.Compression for valid values
		// EFS / NFS-shape Pebble tuning knobs. See db.Options docs
		// for full semantics. 0 = leave Pebble's default for every
		// numeric field here (consistent with the rest of this
		// struct). BytesPerSync also accepts a negative value
		// (db.BytesPerSyncDisabled) to explicitly disable the
		// periodic sync_file_range cadence, which is the
		// EFS-friendly setting.
		TargetFileSizeL0      int64 `yaml:"targetFileSizeL0" default:"0"`      // 0 -> Pebble default 2 MiB
		BytesPerSync          int   `yaml:"bytesPerSync" default:"0"`          // 0 -> Pebble default 512 KiB; <0 -> disabled (EFS-friendly)
		LBaseMaxBytes         int64 `yaml:"lBaseMaxBytes" default:"0"`         // 0 -> Pebble default 64 MiB
		L0StopWritesThreshold int   `yaml:"l0StopWritesThreshold" default:"0"` // 0 -> Pebble default 12
		EnableBloomFilter     bool  `yaml:"enableBloomFilter" default:"false"` // false -> Pebble default (no bloom)
		EnableWAL             bool  `yaml:"enableWAL" default:"false"`
		SyncWrites            bool  `yaml:"syncWrites" default:"false"`
		// PostIngestCompact, when true, triggers a synchronous
		// full-keyspace db.Compact() at the end of ProcessLogs
		// (just before LogProcessor.Finished is set). The
		// compaction collapses L0 sublevel overlap, GC's
		// tombstones, and re-encodes the LSM under the bottom-
		// level compression profile — typically shrinking the
		// on-disk DB by 30-50% and making subsequent indexed
		// range scans (the plugin's hot path) noticeably faster
		// because they touch a single dense level instead of
		// merging across L0/L1/L2. The cost is a one-time
		// post-ingest pause whose duration is bounded by total
		// DB bytes ÷ EFS bandwidth (typically a few minutes on
		// AGI-shaped runs). Default false to preserve the legacy
		// "ingest finished == LogProcessor.Finished == queryable"
		// timing; cmdAgiCreate flips this on for cloud
		// (AWS / GCP) deploys where the post-compaction layout
		// pays for itself immediately on the first plugin query.
		PostIngestCompact bool `yaml:"postIngestCompact" default:"false"`
	} `yaml:"db"`
	// Pipeline tuning. These were under `db:` in earlier versions; they
	// are not engine-level knobs and were moved here for clarity. AGI
	// instances are short-lived and not migrated, so the rename is safe.
	DefaultSetName       string `yaml:"defaultSetName" default:"default"`
	LogFileRangesSetName string `yaml:"logFileRangesSetName" default:"logRanges"`
	TimestampColumnName  string `yaml:"timestampColumnName" default:"timestamp"`
	// MaxPutThreads is the size of the worker goroutine pool that
	// drains resultsChan and forwards rows into the putBatcher
	// shards. Workers do per-row label stamping, missing-bin
	// probe, row materialisation, and submit. They do NOT write
	// to Pebble themselves — that work happens on the batcher
	// shards (see PutBatchShards). Workers are essentially
	// row-prep + submit goroutines.
	//
	// 0 selects auto = clamp(GOMAXPROCS*2, 4, 32). Default 128 on
	// the assumption that a deep pool insulates the upstream
	// resultsChan against single-shard commit-window stalls (each
	// parked worker effectively holds one extra row of in-flight
	// buffer past resultsChan). The cost of the deeper pool is
	// minimal: parked goroutines consume zero CPU, the wakeup
	// path is O(1) regardless of pool size, and 128 stacks add
	// ~256 KiB of resident memory.
	//
	// The auto branch (set this to 0 explicitly) is preserved for
	// constrained deployments that genuinely benefit from a
	// smaller pool — but it is NOT the default because the
	// "deeper pool wastes CPU" hypothesis did not hold up under
	// measurement: in head-to-head pprofs at the same throughput
	// the 128-worker and 16-worker runs had indistinguishable CPU
	// profiles, with the only observable difference being the
	// 16-worker run's smaller resultsChan buffer (now decoupled,
	// so the buffer side is no longer tied to this knob).
	MaxPutThreads int `yaml:"maxPutThreads" default:"128"`
	// PutBatchSize is the number of metric rows accumulated per set
	// before the ingest hot path flushes via db.PutBatch. Larger
	// batches amortise the Pebble.Batch commit overhead and the
	// schema-resolution fast-path's lock churn at the cost of
	// holding more rows in memory between flushes (each row is a
	// few hundred bytes) and a longer worst-case end-to-end
	// latency before a row is queryable.
	//
	// Default 1024. Was 256, raised after pprofs taken with the
	// AssumeNew lock-skip in db.PutBatch showed the pipeline's
	// new gate was per-shard putBatcher.submit blocking — workers
	// stalled because each shard's commit window kept its inCh
	// full. Quadrupling the batch size cuts the number of commit
	// calls per row 4x and gives each shard a longer drain phase
	// between commits, which directly widens the back-pressure
	// window before submit blocks. Per-shard inCh capacity is
	// flushSize*4 = 4096 entries at this default, so a single
	// commit window has to outlast ~4 batches' worth of
	// production before submit becomes the bottleneck.
	//
	// Memory cost at peak (16 shards × 4096 entries × ~150 B/row)
	// is on the order of 10 MiB resident, which is trivial next
	// to the 256 MiB Pebble memtable.
	PutBatchSize int `yaml:"putBatchSize" default:"1024"`
	// PutBatchFlushMs is the maximum age of an in-flight batch
	// before the flusher commits it even if it is below
	// PutBatchSize. Bounds the staleness window between log-line
	// arrival and queryability; 50ms keeps the staleness lower
	// than a typical Grafana refresh interval. Set to 0 to use the
	// 50ms default; very small values (<5ms) defeat the batching
	// benefit because the flusher trips before any meaningful
	// number of rows accumulate.
	PutBatchFlushMs int `yaml:"putBatchFlushMs" default:"50"`
	// PutBatchShards is the number of parallel flusher goroutines
	// behind putBatcher. The pre-sharded batcher used a single
	// flusher whose db.PutBatch throughput became the ingest
	// pipeline's hard ceiling once the upstream metaLock was
	// removed; sharding by maphash(key) lets independent batches
	// commit through Pebble in parallel because db.PutBatch is
	// concurrency-safe and stripedLocks never collide across
	// shards.
	//
	// 0 selects auto = min(GOMAXPROCS, 8). The upper cap reflects
	// that Pebble's commit pipeline saturates well before 8 active
	// writers on typical AGI hardware; raising this knob beyond
	// the available cores adds scheduler churn without raising
	// throughput.
	PutBatchShards int `yaml:"putBatchShards" default:"0"`
	Dedup          struct {
		Enabled   bool `yaml:"enabled" default:"true"`
		ReadBytes int  `yaml:"readBytesCount" default:"1048576"`
	} `yaml:"dedup"`
	Processor struct {
		// MaxConcurrentLogFiles caps how many log files are
		// parsed in parallel (one parser goroutine per file in
		// flight). 0 selects auto = clamp(GOMAXPROCS, 4, 16);
		// the resolution lives in processLogsFeed so the same
		// formula applies whether the value comes from the yaml
		// default, an envvar, or a CLI override. Cloud boxes
		// with 16+ vCPU therefore parse 16 files in parallel by
		// default; small Docker containers (and Docker
		// Desktop's GOMAXPROCS-respected cgroup) get 4-8.
		//
		// 16 is the upper cap because past that the resultsChan
		// buffer (128 slots) and the batcher fan-in saturate;
		// adding more parsers just blocks them on chansend
		// without raising throughput.
		MaxConcurrentLogFiles int `yaml:"maxConcurrentLogFiles" default:"0"`
		LogReadBufferSizeKb   int `yaml:"logReadBufferSizeKb" default:"1024"`
	} `yaml:"processor"`
	PreProcess struct {
		FileThreads         int `yaml:"fileThreads" default:"6"`
		UnpackerFileThreads int `yaml:"unpackerFileThreads" default:"4"`
	} `yaml:"preProcessor"`
	ProgressFile struct {
		DisableWrite   bool          `yaml:"disableWrite" default:"false"`
		OutputFilePath string        `yaml:"outputFilePath" default:"ingest/progress/"`
		WriteInterval  time.Duration `yaml:"writeInterval" default:"10s"`
		Compress       bool          `yaml:"compress" default:"true"`
	} `yaml:"progressFile"`
	ProgressPrint struct {
		Enable               bool          `yaml:"enable" default:"true"`
		UpdateInterval       time.Duration `yaml:"updateInterval" default:"10s"`
		PrintOverallProgress bool          `yaml:"printOverallProgress" default:"true"`
		PrintDetailProgress  bool          `yaml:"printDetailProgress" default:"true"`
	} `yaml:"progressPrint"`
	PatternsFile            string        `yaml:"patternsFile"`
	IngestTimeRanges        TimeRanges    `yaml:"ingestTimeRanges"`
	CollectInfoAsadmTimeout time.Duration `yaml:"collectInfoCommandTimeout" default:"150s"`
	CollectInfoMaxSize      int64         `yaml:"collectInfoMaxSize" default:"20971520"` // files over 20MiB will be considered not collectinfo
	CollectInfoSetName      string        `yaml:"collectInfoSetName" default:"collectinfos"`
	Directories             struct {
		CollectInfo   string `yaml:"collectInfo" default:"ingest/files/collectinfo"`
		Logs          string `yaml:"logs" default:"ingest/files/logs"`
		DirtyTmp      string `yaml:"dirtyTemp" default:"ingest/files/input"`
		NoStatLogs    string `yaml:"noStatOut" default:"ingest/files/logs-cut"`
		OtherFiles    string `yaml:"otherFiles" default:"ingest/files/other"`
		ReadOnlyInput bool   `yaml:"readOnlyInput" default:"false" envconfig:"LOGINGEST_READONLY_INPUT"` // When true, input directory is read-only (e.g., bind mount); copy files instead of moving, don't delete after unpack
	} `yaml:"directories"`
	Downloader struct {
		ConcurrentSources bool        `yaml:"concurrentSources" default:"true"`
		S3Source          *S3Source   `yaml:"s3Source"`
		SftpSource        *SftpSource `yaml:"sftpSource"`
	} `yaml:"downloader"`
	CustomSourceName           string `yaml:"customSourceName" default:"" envconfig:"LOGINGEST_CUSTOM_SRCNAME"`
	FindClusterNameNodeIdRegex string `` /* 143-byte string literal not displayed */

	CPUProfilingOutputFile string `yaml:"cpuProfilingOutputFile" envconfig:"LOGINGEST_CPUPROFILE_FILE"`
	SendClusterInfo        string `yaml:"sendClusterInfo" envconfig:"LOGINGEST_SEND_CLUSTER_INFO"`
	// contains filtered or unexported fields
}

func MakeConfig 

func MakeConfig(setDefaults bool, configFile string, parseEnv bool) (*Config, error)

func MakeConfigReader 

func MakeConfigReader(setDefaults bool, configYaml io.Reader, parseEnv bool) (*Config, error)

type DownloaderFile 

type DownloaderFile struct {
	Size         int64
	LastModified time.Time
	IsDownloaded bool
	Error        string
	StartTime    string
	FinishTime   string
}

type EnumFile 

type EnumFile struct {
	Size int64

	ContentType           string
	IsCollectInfo         bool
	IsArchive             bool
	IsText                bool
	IsTarGz               bool
	IsTarBz               bool
	UnpackFailed          bool
	Errors                []string
	PreProcessDuplicateOf []string
	StartAt               int64 // workaround for log files starting at binary 000s
	PreProcessOutPaths    []string
	// contains filtered or unexported fields
}

type Ingest 

type Ingest struct {
	// contains filtered or unexported fields
}

func Init 

func Init(config *Config) (*Ingest, error)

Init opens its own embedded db handle and initialises the ingest service. Use InitWithDB instead when ingest needs to share a handle with the plugin in the same process — Init's exclusive Pebble lock would otherwise block the co-resident plugin from opening the same directory.

func InitWithDB 

func InitWithDB(config *Config, d *db.DB) (*Ingest, error)

InitWithDB is like Init but uses an externally-owned db handle. The caller retains ownership: Close() on the returned Ingest will NOT close d. This is the entry-point used by the merged agi service (cmdAgiExecService) where ingest and plugin share a single Pebble store opened once at process start.

The caller is responsible for closing d after both ingest and plugin have shut down.

func (*Ingest) Close 

func (i *Ingest) Close()

Close releases the resources owned by this Ingest. It is safe to call multiple times and from concurrent goroutines (e.g. a SIGTERM handler racing the normal-completion deferred call). Close only closes the underlying db handle when ownsDB=true (i.e. Init opened the handle itself); when the handle was injected via InitWithDB, the caller retains ownership and must close it.

func (*Ingest) Download 

func (i *Ingest) Download() error

func (*Ingest) DownloadAsftp 

func (i *Ingest) DownloadAsftp() error

func (*Ingest) DownloadS3 

func (i *Ingest) DownloadS3() error

func (*Ingest) PreProcess 

func (i *Ingest) PreProcess() error

func (*Ingest) ProcessCollectInfo 

func (i *Ingest) ProcessCollectInfo() error

func (*Ingest) ProcessLogs 

func (i *Ingest) ProcessLogs(foundLogs map[string]*LogFile, meta map[string]*metaEntries) error

func (*Ingest) ProcessLogsPrep 

func (i *Ingest) ProcessLogsPrep() (foundLogs map[string]*LogFile, meta map[string]*metaEntries, err error)

func (*Ingest) Unpack 

func (i *Ingest) Unpack() error

type IngestStatusStruct 

type IngestStatusStruct struct {
	Ingest struct {
		Running                  bool
		CompleteSteps            *IngestSteps
		DownloaderCompletePct    int
		DownloaderTotalSize      int64
		DownloaderCompleteSize   int64
		LogProcessorCompletePct  int
		LogProcessorTotalSize    int64
		LogProcessorCompleteSize int64
		Errors                   []string
		ErrorCount               int
	}
	PluginRunning        bool
	GrafanaHelperRunning bool
	System               struct {
		DiskTotalBytes   uint64
		DiskFreeBytes    uint64
		MemoryTotalBytes int
		MemoryFreeBytes  int
	}
}

type IngestSteps 

type IngestSteps struct {
	Init                        bool
	Download                    bool
	Unpack                      bool
	PreProcess                  bool
	ProcessLogs                 bool
	ProcessCollectInfo          bool
	CriticalError               string
	InitStartTime               time.Time
	InitEndTime                 time.Time
	DownloadStartTime           time.Time
	DownloadEndTime             time.Time
	UnpackStartTime             time.Time
	UnpackEndTime               time.Time
	PreProcessStartTime         time.Time
	PreProcessEndTime           time.Time
	ProcessLogsStartTime        time.Time
	ProcessLogsEndTime          time.Time
	ProcessCollectInfoStartTime time.Time
	ProcessCollectInfoEndTime   time.Time
}

type LogFile 

type LogFile struct {
	ClusterName string
	NodePrefix  string
	NodeID      string
	NodeSuffix  string
	Size        int64
	Processed   int64
	Finished    bool
	StartTime   string
	FinishTime  string
}

type MetaEntries 

type MetaEntries map[string]*metaEntries

type NotifyEvent 

type NotifyEvent struct {
	AGIName                    string
	Event                      string
	EventDetail                string
	IsDataInMemory             bool
	IngestStatus               *IngestStatusStruct
	DeploymentJsonGzB64        string
	SSHAuthorizedKeysFileGzB64 string
	Owner                      string
	S3Source                   string
	SftpSource                 string
	LocalSource                string
	Label                      string
}

type Progress 

type Progress struct {
	sync.RWMutex
	Downloader           *ProgressDownloader
	Unpacker             *ProgressUnpacker
	PreProcessor         *ProgressPreProcessor
	LogProcessor         *ProgressLogProcessor
	CollectinfoProcessor *ProgressCollectProcessor
}

type ProgressCollectProcessor 

type ProgressCollectProcessor struct {
	Files    map[string]*CfFile
	Finished bool
	// contains filtered or unexported fields
}

type ProgressDownloader 

type ProgressDownloader struct {
	S3Files   map[string]*DownloaderFile // map[key]*details
	SftpFiles map[string]*DownloaderFile // map[path]*details
	Finished  bool
	// contains filtered or unexported fields
}

type ProgressLogProcessor 

type ProgressLogProcessor struct {
	Files    map[string]*LogFile
	Finished bool

	StartTime  time.Time
	LineErrors *lineErrors
	// contains filtered or unexported fields
}

type ProgressPreProcessor 

type ProgressPreProcessor struct {
	Files                     map[string]*EnumFile // map[path]*details
	CollectInfoUniquePrefixes int
	Finished                  bool

	LastUsedPrefix          int
	LastUsedSuffixForPrefix map[int]int
	NodeToPrefix            map[string]int
	// contains filtered or unexported fields
}

type ProgressUnpacker 

type ProgressUnpacker struct {
	Files    map[string]*EnumFile // map[path]*details
	Finished bool
	// contains filtered or unexported fields
}

type S3Source 

type S3Source struct {
	Enabled     bool   `yaml:"enabled" envconfig:"LOGINGEST_S3SOURCE_ENABLED"`
	Threads     int    `yaml:"threads" envconfig:"LOGINGEST_S3SOURCE_THREADS" default:"4"`
	Region      string `yaml:"region" envconfig:"LOGINGEST_S3SOURCE_REGION"`
	BucketName  string `yaml:"bucketName" envconfig:"LOGINGEST_S3SOURCE_BUCKET"`
	KeyID       string `yaml:"keyID" envconfig:"LOGINGEST_S3SOURCE_KEYID"`
	SecretKey   string `yaml:"secretKey" envconfig:"LOGINGEST_S3SOURCE_SECRET"`
	PathPrefix  string `yaml:"pathPrefix" envconfig:"LOGINGEST_S3SOURCE_PATH"`
	SearchRegex string `yaml:"searchRegex" envconfig:"LOGINGEST_S3SOURCE_REGEX"`
	Endpoint    string `yaml:"endpoint" envconfig:"LOGINGEST_S3SOURCE_ENDPOINT"`
	// contains filtered or unexported fields
}

type SSH 

type SSH struct {
	Ip   string
	User string
	Cert string // key file path
	Pass string // password
	// HostKeyFingerprint is the expected SHA256 host key fingerprint. When set,
	// a server presenting any other key is refused; when empty, the key is
	// accepted unverified and a warning is logged.
	HostKeyFingerprint string
	// contains filtered or unexported fields
}

func (*SSH) Close 

func (ssh_client *SSH) Close()

func (*SSH) Connect 

func (ssh_client *SSH) Connect(mode int) error

type SftpSource 

type SftpSource struct {
	Enabled     bool   `yaml:"enabled" envconfig:"LOGINGEST_SFTPSOURCE_ENABLED"`
	Threads     int    `yaml:"threads" envconfig:"LOGINGEST_SFTPSOURCE_THREADS" default:"4"`
	Host        string `yaml:"host" envconfig:"LOGINGEST_SFTPSOURCE_HOST"`
	Port        int    `yaml:"port" envconfig:"LOGINGEST_SFTPSOURCE_PORT"`
	Username    string `yaml:"username" envconfig:"LOGINGEST_SFTPSOURCE_USER"`
	Password    string `yaml:"password" envconfig:"LOGINGEST_SFTPSOURCE_PASSWORD"`
	KeyFile     string `yaml:"keyFile" envconfig:"LOGINGEST_SFTPSOURCE_KEYFILE"`
	PathPrefix  string `yaml:"pathPrefix" envconfig:"LOGINGEST_SFTPSOURCE_PATH"`
	SearchRegex string `yaml:"searchRegex" envconfig:"LOGINGEST_SFTPSOURCE_REGEX"`
	// HostKeyFingerprint is the server's SHA256 host key fingerprint, captured
	// by 'aerolab agi create' when it validated the SFTP credentials. Ingest
	// refuses to connect if the server later presents a different key, so the
	// SFTP password cannot be harvested by an intercepting host. Empty means
	// the source predates this check and the key is not verified.
	HostKeyFingerprint string `yaml:"hostKeyFingerprint" envconfig:"LOGINGEST_SFTPSOURCE_HOSTKEY"`
	// contains filtered or unexported fields
}

type TimeRanges 

type TimeRanges struct {
	Enabled bool      `yaml:"enabled" envconfig:"LOGINGEST_TIMERANGE_ENABLE" default:"false"`
	From    time.Time `yaml:"from" envconfig:"LOGINGEST_TIMERANGE_FROM"`
	To      time.Time `yaml:"to" envconfig:"LOGINGEST_TIMERANGE_TO"`
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.