ingest

package module
v3.3.12 Latest Latest
Warning

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

 Go to latest Published: May 13, 2020 License: BSD-2-Clause Imports: 28 Imported by: 36

README

The Gravwell Ingest API

API documentation: godoc.org/github.com/gravwell/ingest

This package provides methods to build ingesters for the Gravwell analytics platform. Ingesters take raw data from a particular source (pcap, log files, a camera, etc.), bundle it into Entries, and ship the Entries up to the Gravwell indexers.

An Entry (defined in the sub-package github.com/gravwell/ingest/entry) looks like this:

type Entry struct {
    TS   Timestamp
    SRC  net.IP
    Tag  EntryTag
    Data []byte
}

The most important element is Data; this is simply a discrete piece of information you want to store as an entry, be it binary or text. The Tag field associates the entry with a specific tag in the indexer, making it easier to search later. The timestamp and source IP give additional information about the entry.

There are two ways to actually get Entries into Gravwell: the IngestConnection and the IngestMuxer. An IngestConnection is a connection to a single destination, while an IngestMuxer can connect to multiple destinations simultaneously to improve ingestion rate.

The example below shows the basics of getting Entries into Gravwell using an IngestConnection (the simpler method). See ingester_example for a more detailed (and functional) example using an IngestMuxer, or see github.com/gravwell/ingesters for open-source real-world examples.

package main

import (
	"github.com/gravwell/ingest/v3"
	"github.com/gravwell/ingest/v3/entry"
	"log"
	"net"
)

func main() {
	// Get an IngestConnection to localhost, using the shared secret "IngestSecrets"
	// and specifying that we'll be using the tag "test-tag"
	igst, err := ingest.InitializeConnection("tcp://127.0.0.1:4023", "IngestSecrets",[]string{"testtag"}, "", "", false)
	if err != nil {
		log.Fatalf("Couldn't open connection to ingester: %v", err)
	}
	defer igst.Close()

	// We need to get the numeric value for the tag we're using
	tagid, ok := igst.GetTag("testtag")
	if !ok {
		log.Fatal("couldn't look up tag")
	}

	// Now we'll create an Entry
	ent := entry.Entry{
		TS: entry.Now(),
		SRC: net.ParseIP("127.0.0.1"),
		Tag: tagid,
		Data: []byte("This is my test data!"),
	}

	// And finally write the Entry
	igst.WriteEntry(&ent)
}

Documentation

Overview

Example (Simplest)

SimplestExample is the simplest possible example of ingesting a single Entry. 

/*************************************************************************
 * Copyright 2017 Gravwell, Inc. All rights reserved.
 * Contact: <legal@gravwell.io>
 *
 * This software may be modified and distributed under the terms of the
 * BSD 2-clause license. See the LICENSE file for details.
 **************************************************************************/

package main

import (
	"log"
	"net"

	"github.com/gravwell/ingest/v3"
	"github.com/gravwell/ingest/v3/entry"
)

var (
	dst          = "tcp://127.0.0.1:4023"
	sharedSecret = "IngestSecrets"
	simple_tags  = []string{"testtag"}
)

// SimplestExample is the simplest possible example of ingesting a single Entry.
func main() {
	// Get an IngestConnection
	igst, err := ingest.InitializeConnection(dst, sharedSecret, simple_tags, "", "", false)
	if err != nil {
		log.Fatalf("Couldn't open connection to ingester: %v", err)
	}
	defer igst.Close()

	// We need to get the numeric value for the tag we're using
	tagid, ok := igst.GetTag(simple_tags[0])
	if !ok {
		log.Fatal("couldn't look up tag")
	}

	// Now we'll create an Entry
	ent := entry.Entry{
		TS:   entry.Now(),
		SRC:  net.ParseIP("127.0.0.1"),
		Tag:  tagid,
		Data: []byte("This is my test data!"),
	}

	// And finally write the Entry
	igst.WriteEntry(&ent)
}

Output:

Index

Examples

Constants

View Source 
const (
	//MAJOR API VERSIONS should always be compatible, there just may be additional features
	API_VERSION_MAJOR uint32 = 0
	API_VERSION_MINOR uint32 = 4
)
View Source 
const (
	// The number of times to hash the shared secret
	HASH_ITERATIONS uint16 = 16
	// Auth protocol version number
	VERSION uint16 = 0x4
	// Authenticated, but not ready for ingest
	STATE_AUTHENTICATED uint32 = 0xBEEF42
	// Not authenticated
	STATE_NOT_AUTHENTICATED uint32 = 0xFEED51
	// Authenticated and ready for ingest
	STATE_HOT uint32 = 0xCAFE54
)
View Source 
const (
	READ_BUFFER_SIZE int = 4 * 1024 * 1024
	//TODO - we should really discover the MTU of the link and use that
	ACK_WRITER_BUFFER_SIZE int = (ackEncodeSize * MAX_UNCONFIRMED_COUNT)
	ACK_WRITER_CAN_WRITE   int = (ackEncodeSize * (MAX_UNCONFIRMED_COUNT / 2))
)
View Source 
const (
	ACK_SIZE int = 12 //ackmagic + entrySendID
	//READ_ENTRY_HEADER_SIZE should be 46 bytes
	//34 + 4 + 4 + 8 (magic, data len, entry ID)
	READ_ENTRY_HEADER_SIZE int = entry.ENTRY_HEADER_SIZE + 12
	//TODO: We should make this configurable by configuration
	MAX_ENTRY_SIZE              int           = 128 * 1024 * 1024
	WRITE_BUFFER_SIZE           int           = 1024 * 1024
	MAX_WRITE_ERROR             int           = 4
	BUFFERED_ACK_READER_SIZE    int           = ACK_SIZE * MAX_UNCONFIRMED_COUNT
	CLOSING_SERVICE_ACK_TIMEOUT time.Duration = 3 * time.Second

	MAX_UNCONFIRMED_COUNT int = 1024 * 4

	MINIMUM_TAG_RENEGOTIATE_VERSION uint16 = 0x2 // minimum server version to renegotiate tags
	MINIMUM_ID_VERSION              uint16 = 0x3 // minimum server version to send ID info
	MINIMUM_INGEST_OK_VERSION       uint16 = 0x4 // minimum server version to ask

)
View Source 
const (
	KB       uint64  = 1024
	MB       uint64  = 1024 * KB
	GB       uint64  = 1024 * MB
	TB       uint64  = 1024 * GB
	PB       uint64  = 1024 * TB
	YB       uint64  = 1024 * PB
	K                = 1000.0
	M                = K * 1000.0
	G                = M * 1000.0
	T                = G * 1000.0
	P                = G * 1000.0
	Y                = P * 1000.0
	NsPerSec float64 = 1000000000.0
)
View Source 
const (
	MIN_REMOTE_KEYSIZE int           = 16
	DIAL_TIMEOUT       time.Duration = (1 * time.Second)
	CHANNEL_BUFFER     int           = 4096
	DEFAULT_TLS_PORT   int           = 4024
	DEFAULT_CLEAR_PORT int           = 4023
	DEFAULT_PIPE_PATH  string        = "/opt/gravwell/comms/pipe"
	FORBIDDEN_TAG_SET  string        = "!@#$%^&*()=+<>,.:;`\"'{[}]|\\ 	" // includes space and tab characters at the end
)
View Source 
const (
	//This MUST be > 1 or the universe explodes
	//no matter what a user requests, this is the maximum
	//basically a sanity check
	ABSOLUTE_MAX_UNCONFIRMED_WRITES int = 0xffff
)

Variables

View Source 
var (
	ErrInvalidStateResponseLen = errors.New("Invalid state response length")
	ErrInvalidTagRequestLen    = errors.New("Invalid tag request length")
	ErrInvalidTagResponseLen   = errors.New("Invalid tag response length")
	ErrFailedAuthHashGen       = errors.New("Failed to generate authentication hash")
	ErrFailedAuth              = errors.New("Failed authentication, bad secret")
	ErrFailedTagNegotiation    = errors.New("Failed to negotiate tags")
)
View Source 
var (
	ErrActiveHotBlocks        = errors.New("There are active hotblocks, close pitched data")
	ErrNoActiveDB             = errors.New("No active database")
	ErrNoKey                  = errors.New("No key set on entry block")
	ErrCacheAlreadyRunning    = errors.New("Cache already running")
	ErrCacheNotRunning        = errors.New("Cache is not running")
	ErrBucketMissing          = errors.New("Cache bucket is missing")
	ErrCannotSyncWhileRunning = errors.New("Cannot sync while running")
	ErrCannotPopWhileRunning  = errors.New("Cannot pop while running")
	ErrInvalidKey             = errors.New("Key bytes are invalid")
	ErrBoltLockFailed         = errors.New("Failed to acquire lock for ingest cache.  The file is locked by another process")
)
View Source 
var (
	ErrEmptyTag     = errors.New("Tag name is empty")
	ErrForbiddenTag = errors.New("Forbidden character in tag")
)
View Source 
var (
	ErrAllConnsDown          = errors.New("All connections down")
	ErrNotRunning            = errors.New("Not running")
	ErrNotReady              = errors.New("Not ready to start")
	ErrTagNotFound           = errors.New("Tag not found")
	ErrTagMapInvalid         = errors.New("Tag map invalid")
	ErrNoTargets             = errors.New("No connections specified")
	ErrConnectionTimeout     = errors.New("Connection timeout")
	ErrSyncTimeout           = errors.New("Sync timeout")
	ErrEmptyAuth             = errors.New("Ingest key is empty")
	ErrEmergencyListOverflow = errors.New("Emergency list overflow")
	ErrTimeout               = errors.New("Timed out waiting for ingesters")
	ErrWriteTimeout          = errors.New("Timed out waiting to write entry")
)
View Source 
var (
	ErrFailedParseLocalIP    = errors.New("Failed to parse the local address")
	ErrMalformedDestination  = errors.New("Malformed destination string")
	ErrInvalidCerts          = errors.New("Failed to get certificates")
	ErrInvalidDest           = errors.New("Invalid destination")
	ErrInvalidRemoteKeySize  = errors.New("Invalid remote keysize")
	ErrInvalidConnectionType = errors.New("Invalid connection type")
	ErrInvalidSecret         = errors.New("Failed to login, invalid secret")
)

Functions

func CheckTag 

func CheckTag(tag string) error

CheckTag takes a tag name and returns an error if it contains any characters which are not allowed in tags.

func ConnectionType 

func ConnectionType(dst string) (string, string, error)

ConnectionType cracks out the type of connection and returns its type, the target, and/or an error

func HumanCount 

func HumanCount(b uint64) string

HumanCount will take a number and return an appropriately-scaled string, e.g. HumanCount(12500) will return "12.50 K"

func HumanEntryRate 

func HumanEntryRate(b uint64, dur time.Duration) string

HumanEntryRate will take an entry count and duration and produce a human readable string in terms of entries per second. e.g. 2400 K entries /s

func HumanLineRate 

func HumanLineRate(b uint64, dur time.Duration) string

HumanLineRate will take a byte count and duration and produce a human readable string in terms of bits. e.g. Megabits/s (Mbps)

func HumanRate 

func HumanRate(b uint64, dur time.Duration) string

HumanRate will take a byte count and duration and produce a human readable string showing data per second. e.g. Megabytes/s (MB/s)

func HumanSize 

func HumanSize(b uint64) string

HumanSize will take a byte count and duration and produce a human readable string showing data per second. e.g. Megabytes/s (MB/s)

func PrintVersion 

func PrintVersion(wtr io.Writer)

func VerifyResponse 

func VerifyResponse(auth AuthHash, chal Challenge, resp ChallengeResponse) error

VerifyResponse takes a hash and challenge and computes a completed response. If the computed response does not match an error is returned. If the response matches, nil is returned

Types

type AuthHash 

type AuthHash [16]byte

AuthHash represents a hashed shared secret.

func GenAuthHash 

func GenAuthHash(password string) (AuthHash, error)

GenAuthHash takes a key and generates a hash using the "password" token we iterate over the value, hashing with MD5 and SHA256. We choose these two algorithms because they aren't too heavy, but the alternating makes it very difficult to optimize in an FPGA or ASIC.

type Challenge 

type Challenge struct {
	// Number of times to iterate the hash
	Iterate uint16
	// The random number to be hashed with the secret
	RandChallenge [32]byte
	// Authentication version number
	Version uint16
}

Challenge request, used to validate remote clients. The server generates RandChallenge, a random number which is hashed with the pre-hashed shared secret, then run through Iterate iterations of md5 and sha256 to create the response.

func NewChallenge 

func NewChallenge(auth AuthHash) (Challenge, error)

NewChallenge generates a random hash string and a random iteration count

func (*Challenge) Read 

func (c *Challenge) Read(r io.Reader) error

Read the challenge from reader

func (*Challenge) Write 

func (c *Challenge) Write(w io.Writer) error

Write out the challenge to w

type ChallengeResponse 

type ChallengeResponse struct {
	Response [32]byte
}

ChallengeResponse is the resulting hash sent back as part of the challenge/response process.

func GenerateResponse 

func GenerateResponse(auth AuthHash, ch Challenge) (*ChallengeResponse, error)

GenerateResponse creates a ChallengeResponse based on the Challenge and AuthHash

func (*ChallengeResponse) Read 

func (cr *ChallengeResponse) Read(r io.Reader) error

Decode the ChallengeResponse from the reader

func (*ChallengeResponse) Write 

func (cr *ChallengeResponse) Write(w io.Writer) error

Write the challenge response to the writer

type EntryReader 

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

func NewEntryReader 

func NewEntryReader(conn net.Conn) (*EntryReader, error)

func NewEntryReaderEx 

func NewEntryReaderEx(cfg EntryReaderWriterConfig) (*EntryReader, error)

func (*EntryReader) Close 

func (er *EntryReader) Close() error

func (*EntryReader) GetIngesterAPIVersion added in v3.3.6 

func (er *EntryReader) GetIngesterAPIVersion() uint16

func (*EntryReader) GetIngesterInfo added in v3.2.4 

func (er *EntryReader) GetIngesterInfo() (string, string, string)

func (*EntryReader) IngestOK added in v3.3.6 

func (er *EntryReader) IngestOK(ok bool) (err error)

IngestOK waits for the ingester to send an INGEST_OK message and responds with the argument given. Any other command will make it return an error

func (*EntryReader) Read 

func (er *EntryReader) Read() (e *entry.Entry, err error)

func (*EntryReader) SendThrottle 

func (er *EntryReader) SendThrottle(d time.Duration) error

func (*EntryReader) SetTagManager 

func (er *EntryReader) SetTagManager(tm TagManager)

SetTagManager gives a handle on the instantiator's tag management system. If this is not set, tags cannot be negotiated on the fly

func (*EntryReader) SetupConnection added in v3.3.6 

func (er *EntryReader) SetupConnection() (err error)

SetupConnection negotiations ingester API version and other information It should properly handle old ingesters, too

func (*EntryReader) Start 

func (er *EntryReader) Start() error

type EntryReaderWriterConfig 

type EntryReaderWriterConfig struct {
	Conn                  net.Conn
	OutstandingEntryCount int
	BufferSize            int
	Timeout               time.Duration
	TagMan                TagManager
}

type EntryWriter 

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

func NewEntryWriter 

func NewEntryWriter(conn net.Conn) (*EntryWriter, error)

func NewEntryWriterEx 

func NewEntryWriterEx(cfg EntryReaderWriterConfig) (*EntryWriter, error)

func (*EntryWriter) Ack 

func (ew *EntryWriter) Ack() error

Ack will block waiting for at least one ack to free up a slot for sending

func (*EntryWriter) Close 

func (ew *EntryWriter) Close() (err error)

func (*EntryWriter) ForceAck 

func (ew *EntryWriter) ForceAck() error

func (*EntryWriter) IdentifyIngester added in v3.2.4 

func (ew *EntryWriter) IdentifyIngester(name string, version string, id string) (err error)

func (*EntryWriter) IngestOK added in v3.3.6 

func (ew *EntryWriter) IngestOK() (ok bool, err error)

func (*EntryWriter) NegotiateTag 

func (ew *EntryWriter) NegotiateTag(name string) (tg entry.EntryTag, err error)

func (*EntryWriter) OpenSlots 

func (ew *EntryWriter) OpenSlots(ent *entry.Entry) int

OpenSlots informs the caller how many slots are available before we must service acks. This is used for mostly in a multiplexing system where we want to know how much we can write before we need to service acks and move on.

func (EntryWriter) OptimalBatchWriteSize 

func (ew EntryWriter) OptimalBatchWriteSize() int

func (*EntryWriter) OverrideAckTimeout 

func (ew *EntryWriter) OverrideAckTimeout(t time.Duration) error

func (*EntryWriter) Ping 

func (ew *EntryWriter) Ping() (err error)

Ping is essentially a force ack, we send a PING command, which will cause the server to flush all acks and a PONG command. We read until we get the PONG

func (*EntryWriter) SendIngesterAPIVersion added in v3.3.6 

func (ew *EntryWriter) SendIngesterAPIVersion() (err error)

func (*EntryWriter) SetConn 

func (ew *EntryWriter) SetConn(c conn)

func (*EntryWriter) Write 

func (ew *EntryWriter) Write(ent *entry.Entry) error

Write expects to have exclusive control over the entry and all its buffers from the period of write and forever after. This is because it needs to be able to resend the entry if it fails to confirm. If a buffer is re-used and the entry fails to confirm we will send the new modified buffer which may not have the original data.

func (*EntryWriter) WriteBatch 

func (ew *EntryWriter) WriteBatch(ents [](*entry.Entry)) error

WriteBatch takes a slice of entries and writes them, this function is useful in multithreaded environments where we want to lessen the impact of hits on a channel by threads

func (*EntryWriter) WriteSync 

func (ew *EntryWriter) WriteSync(ent *entry.Entry) error

func (*EntryWriter) WriteWithHint 

func (ew *EntryWriter) WriteWithHint(ent *entry.Entry) (bool, error)

WriteWithHint behaves exactly like Write but also returns a bool which indicates whether or not the a flush was required. This function method is primarily used when muxing across multiple indexers, so the muxer knows when to transition to the next indexer

type IngestCache 

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

func NewIngestCache 

func NewIngestCache(c IngestCacheConfig) (*IngestCache, error)

NewIngestCache creates a ingest cache and gets a handle on the store if specified. If the store can't be opened when asked for, we return an error

func (*IngestCache) Close 

func (ic *IngestCache) Close() error

Close flushes hot blocks to the store and closes the cache

func (*IngestCache) Count 

func (ic *IngestCache) Count() uint64

Count returns the number of entries held, including in the storage system

func (*IngestCache) GetTagList 

func (ic *IngestCache) GetTagList() ([]string, error)

func (*IngestCache) HotBlocks 

func (ic *IngestCache) HotBlocks() int

HotBlocks returns the number of blocks currently held in memory

func (*IngestCache) MemoryCacheSize 

func (ic *IngestCache) MemoryCacheSize() uint64

MemoryCacheSize returns how much is held in memory

func (*IngestCache) PopBlock 

func (ic *IngestCache) PopBlock() (*entry.EntryBlock, error)

PopBlock will pop any available block and hand it back the block is entirely removed from the cache

func (*IngestCache) Start 

func (ic *IngestCache) Start(eChan chan *entry.Entry, bChan chan []*entry.Entry) error

func (*IngestCache) Stop 

func (ic *IngestCache) Stop() error

func (*IngestCache) StoredBlocks 

func (ic *IngestCache) StoredBlocks() int

StoredBlocks returns the total number of blocks held in the store

func (*IngestCache) Sync 

func (ic *IngestCache) Sync() error

Sync flushes all hot blocks to the data store. If we an in memory only cache then we throw an error

func (*IngestCache) UpdateStoredTagList 

func (ic *IngestCache) UpdateStoredTagList(tags []string) error

type IngestCacheConfig 

type IngestCacheConfig struct {
	FileBackingLocation string
	TickInterval        time.Duration
	MemoryCacheSize     uint64
	MaxCacheSize        uint64
}

type IngestCommand 

type IngestCommand uint32
const (
	//ingester commands
	INVALID_MAGIC           IngestCommand = 0x00000000
	NEW_ENTRY_MAGIC         IngestCommand = 0xC7C95ACB
	FORCE_ACK_MAGIC         IngestCommand = 0x1ADF7350
	CONFIRM_ENTRY_MAGIC     IngestCommand = 0xF6E0307E
	THROTTLE_MAGIC          IngestCommand = 0xBDEACC1E
	PING_MAGIC              IngestCommand = 0x88770001
	PONG_MAGIC              IngestCommand = 0x88770008
	TAG_MAGIC               IngestCommand = 0x18675300
	CONFIRM_TAG_MAGIC       IngestCommand = 0x18675301
	ERROR_TAG_MAGIC         IngestCommand = 0x18675302
	ID_MAGIC                IngestCommand = 0x22793400
	CONFIRM_ID_MAGIC        IngestCommand = 0x22793401
	API_VER_MAGIC           IngestCommand = 0x22334400
	CONFIRM_API_VER_MAGIC   IngestCommand = 0x22334401
	INGEST_OK_MAGIC         IngestCommand = 0x33445500
	CONFIRM_INGEST_OK_MAGIC IngestCommand = 0x33445501
)

func (IngestCommand) Buff 

func (ic IngestCommand) Buff() (b []byte)

func (IngestCommand) String 

func (ic IngestCommand) String() string

type IngestConnection deprecated 

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

IngestConnection is a lower-level interface for connecting to and communicating with a single indexer. It is kept public for compatibility, but should not be used in new projects.

Deprecated: Use the IngestMuxer instead.

func InitializeConnection deprecated 

func InitializeConnection(dst, authString string, tags []string, pubKey, privKey string, verifyRemoteKey bool) (*IngestConnection, error)

InitializeConnection is a simple wrapper to get a line to an ingester. callers can just call this function and get back a hot ingest connection We take care of establishing the connection and shuttling auth around

Deprecated: Use the IngestMuxer instead.

func NewPipeConnection deprecated 

func NewPipeConnection(dst string, auth AuthHash, tags []string) (*IngestConnection, error)

This function will create a new NamedPipe connection to a local system. We have NO WAY of knowing which process is REALLY on the othe other end of the pipe. But it is assumed that gravwell will be running with highly limited priveleges, so if the integrity of the local system is compromised, its already over.

Deprecated: Use the IngestMuxer instead.

func NewTCPConnection deprecated 

func NewTCPConnection(dst string, auth AuthHash, tags []string) (*IngestConnection, error)

This function will create a new cleartext TCP connection to a remote system. No verification of the server is performed AT ALL. All traffic is snoopable and modifiable. If someone has control of the network, they will be able to inject and monitor this traffic.

dst: should be a address:port pair. For example "ingest.gravwell.com:4042" or "10.0.0.1:4042"

Deprecated: Use the IngestMuxer instead.

func NewTLSConnection deprecated 

func NewTLSConnection(dst string, auth AuthHash, certs *TLSCerts, verify bool, tags []string) (*IngestConnection, error)

NewTLSConnection will create a new connection to a remote system using a secure TLS tunnel. If the remotePubKey in TLSCerts is set, we will verify the public key of the remote server and bail if it doesn't match. This is a basic MitM protection. This requires that we HAVE the remote public key, getting that will be done else where.

Deprecated: Use the IngestMuxer instead.

func (*IngestConnection) Close 

func (igst *IngestConnection) Close() error

func (*IngestConnection) GetTag 

func (igst *IngestConnection) GetTag(name string) (entry.EntryTag, bool)

func (*IngestConnection) IdentifyIngester added in v3.2.4 

func (igst *IngestConnection) IdentifyIngester(name, version, id string) (err error)

func (*IngestConnection) IngestOK added in v3.3.6 

func (igst *IngestConnection) IngestOK() (ok bool, err error)

IngestOK asks the indexer if it is ok to start sending entries yet.

func (*IngestConnection) NegotiateTag 

func (igst *IngestConnection) NegotiateTag(name string) (tg entry.EntryTag, err error)

func (*IngestConnection) Running 

func (igst *IngestConnection) Running() bool

func (*IngestConnection) Source 

func (igst *IngestConnection) Source() (net.IP, error)

func (*IngestConnection) String added in v3.3.3 

func (igst *IngestConnection) String() (s string)

func (*IngestConnection) Sync 

func (igst *IngestConnection) Sync() error
Sync causes the entry writer to force an ack from teh server.  This ensures that all

* entries that have been written are flushed and fully acked by the server.

func (*IngestConnection) Write 

func (igst *IngestConnection) Write(ts entry.Timestamp, tag entry.EntryTag, data []byte) error

func (*IngestConnection) WriteBatchEntry 

func (igst *IngestConnection) WriteBatchEntry(ents []*entry.Entry) error

WriteBatchEntry DOES NOT populate the source on write, the caller must do so

func (*IngestConnection) WriteEntry 

func (igst *IngestConnection) WriteEntry(ent *entry.Entry) error

func (*IngestConnection) WriteEntrySync 

func (igst *IngestConnection) WriteEntrySync(ent *entry.Entry) error

type IngestLogger 

type IngestLogger interface {
	Error(string, ...interface{}) error
	Warn(string, ...interface{}) error
	Info(string, ...interface{}) error
}

func NoLogger 

func NoLogger() IngestLogger

type IngestMuxer 

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

func NewIngestMuxer 

func NewIngestMuxer(dests []Target, tags []string, pubKey, privKey string) (*IngestMuxer, error)

func NewIngestMuxerExt 

func NewIngestMuxerExt(dests []Target, tags []string, pubKey, privKey string, chanSize int) (*IngestMuxer, error)

func NewMuxer 

func NewMuxer(c MuxerConfig) (*IngestMuxer, error)

func NewUniformIngestMuxer 

func NewUniformIngestMuxer(dests, tags []string, authString, pubKey, privKey, remoteKey string) (*IngestMuxer, error)

NewIngestMuxer creates a new muxer that will automatically distribute entries amongst the clients

func NewUniformIngestMuxerExt 

func NewUniformIngestMuxerExt(dests, tags []string, authString, pubKey, privKey, remoteKey string, chanSize int) (*IngestMuxer, error)

func NewUniformMuxer 

func NewUniformMuxer(c UniformMuxerConfig) (*IngestMuxer, error)

func (*IngestMuxer) Close 

func (im *IngestMuxer) Close() error

Close the connection

func (*IngestMuxer) Dead 

func (im *IngestMuxer) Dead() (int, error)

Dead returns how many connections are currently dead

func (*IngestMuxer) Error 

func (im *IngestMuxer) Error(format string, args ...interface{}) error

GravwellError send an error entry down the line with the gravwell tag

func (*IngestMuxer) GetTag 

func (im *IngestMuxer) GetTag(tag string) (tg entry.EntryTag, err error)

GetTag pulls back an intermediary tag id the intermediary tag has NO RELATION to the backend servers tag mapping it is used to speed along tag mappings

func (*IngestMuxer) Hot 

func (im *IngestMuxer) Hot() (int, error)

Hot returns how many connections are functioning

func (*IngestMuxer) Info 

func (im *IngestMuxer) Info(format string, args ...interface{}) error

func (*IngestMuxer) LookupTag added in v3.3.8 

func (im *IngestMuxer) LookupTag(tg entry.EntryTag) (name string, ok bool)

LookupTag will reverse a tag id into a name, this operation is more expensive than a straight lookup Users that expect to translate a tag repeatedly should maintain their own tag map

func (*IngestMuxer) NegotiateTag 

func (im *IngestMuxer) NegotiateTag(name string) (tg entry.EntryTag, err error)

NegotiateTag will attempt to lookup a tag name in the negotiated set The the tag name has not already been negotiated, the muxer will contact each indexer and negotiate it. This call can potentially block and fail

func (*IngestMuxer) Size 

func (im *IngestMuxer) Size() (int, error)

Size returns the total number of specified connections, hot or dead

func (*IngestMuxer) SourceIP 

func (im *IngestMuxer) SourceIP() (net.IP, error)

SourceIP is a convienence function used to pull back a source value

func (*IngestMuxer) Start 

func (im *IngestMuxer) Start() error

Start starts the connection process. This will return immediately, and does not mean that connections are ready. Callers should call WaitForHot immediately after to wait for the connections to be ready.

func (*IngestMuxer) Sync 

func (im *IngestMuxer) Sync(to time.Duration) error

func (*IngestMuxer) SyncContext 

func (im *IngestMuxer) SyncContext(ctx context.Context, to time.Duration) error

func (*IngestMuxer) WaitForHot 

func (im *IngestMuxer) WaitForHot(to time.Duration) error

WaitForHot waits until at least one connection goes into the hot state The timout duration parameter is an optional timeout, if zero, it waits indefinitely

func (*IngestMuxer) WaitForHotContext 

func (im *IngestMuxer) WaitForHotContext(ctx context.Context, to time.Duration) error

func (*IngestMuxer) Warn 

func (im *IngestMuxer) Warn(format string, args ...interface{}) error

func (*IngestMuxer) Write 

func (im *IngestMuxer) Write(tm entry.Timestamp, tag entry.EntryTag, data []byte) error

Write puts together the arguments to create an entry and writes it to the queue to be sent out by the first available entry writer routine, if all routines are dead, THIS WILL BLOCK once the channel fills up. We figure this is a natural "wait" mechanism

func (*IngestMuxer) WriteBatch 

func (im *IngestMuxer) WriteBatch(b []*entry.Entry) error

WriteBatch puts a slice of entries into the queue to be sent out by the first available entry writer routine. The entry writer routines will consume the entire slice, so extremely large slices will go to a single indexer.

func (*IngestMuxer) WriteBatchContext 

func (im *IngestMuxer) WriteBatchContext(ctx context.Context, b []*entry.Entry) error

WriteBatchContext puts a slice of entries into the queue to be sent out by the first available entry writer routine. The entry writer routines will consume the entire slice, so extremely large slices will go to a single indexer. if a cancellation context isn't needed, use WriteBatch

func (*IngestMuxer) WriteContext 

func (im *IngestMuxer) WriteContext(ctx context.Context, tm entry.Timestamp, tag entry.EntryTag, data []byte) error

WriteContext puts together the arguments to create an entry and writes it to the queue to be sent out by the first available entry writer routine, if all routines are dead, THIS WILL BLOCK once the channel fills up. We figure this is a natural "wait" mechanism if the context isn't needed use Write instead

func (*IngestMuxer) WriteEntry 

func (im *IngestMuxer) WriteEntry(e *entry.Entry) error

WriteEntry puts an entry into the queue to be sent out by the first available entry writer routine, if all routines are dead, THIS WILL BLOCK once the channel fills up. We figure this is a natural "wait" mechanism

func (*IngestMuxer) WriteEntryContext 

func (im *IngestMuxer) WriteEntryContext(ctx context.Context, e *entry.Entry) error

WriteEntryContext puts an entry into the queue to be sent out by the first available entry writer routine, if all routines are dead, THIS WILL BLOCK once the channel fills up. We figure this is a natural "wait" mechanism if not using a context, use WriteEntry as it is faster due to the lack of a select

func (*IngestMuxer) WriteEntryTimeout 

func (im *IngestMuxer) WriteEntryTimeout(e *entry.Entry, d time.Duration) (err error)

WriteEntryAttempt attempts to put an entry into the queue to be sent out of the first available writer routine. This write is opportunistic and contains a timeout. It is therefor every expensive and shouldn't be used for normal writes The typical use case is via the gravwell_log calls

type Logger 

type Logger interface {
	Info(string, ...interface{}) error
	Warn(string, ...interface{}) error
	Error(string, ...interface{}) error
	InfoWithDepth(int, string, ...interface{}) error
	WarnWithDepth(int, string, ...interface{}) error
	ErrorWithDepth(int, string, ...interface{}) error
}

type MuxerConfig 

type MuxerConfig struct {
	Destinations    []Target
	Tags            []string
	PublicKey       string
	PrivateKey      string
	VerifyCert      bool
	ChannelSize     int
	EnableCache     bool
	CacheConfig     IngestCacheConfig
	LogLevel        string
	Logger          Logger
	IngesterName    string
	IngesterVersion string
	IngesterUUID    string
	RateLimitBps    int64
}

type StateResponse 

type StateResponse struct {
	ID   uint32
	Info string
}

StateResponse

func (*StateResponse) Read 

func (sr *StateResponse) Read(r io.Reader) error

Read reads a state response from the reader

func (*StateResponse) Write 

func (sr *StateResponse) Write(w io.Writer) error

Write the StateResponse

type TLSCerts 

type TLSCerts struct {
	Cert tls.Certificate
}

type TagManager 

type TagManager interface {
	GetAndPopulate(string) (entry.EntryTag, error)
}

type TagRequest 

type TagRequest struct {
	Count uint32
	Tags  []string
}

TagRequest is used to request tags for the ingester

func (*TagRequest) Read 

func (tr *TagRequest) Read(r io.Reader) error

Read a TagRequest structure

func (*TagRequest) Write 

func (tr *TagRequest) Write(w io.Writer) error

Write a TagRequest

type TagResponse 

type TagResponse struct {
	Count uint32
	Tags  map[string]entry.EntryTag
}

TagResponse represents the Tag Name to Tag Number mapping supported by the ingest server

func (*TagResponse) Read 

func (tr *TagResponse) Read(r io.Reader) error

Read a TagResponse

func (*TagResponse) Write 

func (tr *TagResponse) Write(w io.Writer) error

Write a TagResponse

type Target 

type Target struct {
	Address string
	Secret  string
}

type TargetError 

type TargetError struct {
	Address string
	Error   error
}

type UniformMuxerConfig 

type UniformMuxerConfig struct {
	Destinations    []string
	Tags            []string
	Auth            string
	PublicKey       string
	PrivateKey      string
	VerifyCert      bool
	ChannelSize     int
	EnableCache     bool
	CacheConfig     IngestCacheConfig
	LogLevel        string
	Logger          Logger
	IngesterName    string
	IngesterVersion string
	IngesterUUID    string
	RateLimitBps    int64
}

Source Files

View all Source files

Directories

Path Synopsis
Package config provides a common base for Gravwell ingester config files.
 Package config provides a common base for Gravwell ingester config files.
Package processors implements preprocessors for ingesters.
 Package processors implements preprocessors for ingesters.

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.