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
}

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