s3batcher

Documentation

Overview

# The MIT License (MIT) # # Copyright (c) 2019 Carbon Black # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in all # copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE.

Index

• func FlushToS3(s *S3BufferClient, s3u s3manageriface.UploaderAPI) error
• func Get_Key_Prefix() string
• func PutRecordInBuffer(s *S3BufferClient, record []byte, seq Sequence, isOverflowData bool) (bool, error)
• func UploadToS3(s3u s3manageriface.UploaderAPI, file io.Reader, bucket string, key string) error
• type Buffer
  • func (b *Buffer) LockBuffer()
  • func (b *Buffer) Read(p []byte) (n int, err error)
  • func (b *Buffer) ReadFrom(r io.Reader) (n int64, err error)
  • func (b *Buffer) Reset()
  • func (b *Buffer) String() string
  • func (b *Buffer) UnlockBuffer()
  • func (b *Buffer) Write(p []byte) (n int, err error)
• type S3BufferClient
  • func NewBufferClient(Env string, Shard_Id string, S3_Bucket string, Region string, ...) (*S3BufferClient, error)
  • func (s *S3BufferClient) Background_Worker_Is_Dead() bool
  • func (s *S3BufferClient) Begin_Background_Worker(s3u s3manageriface.UploaderAPI)
  • func (s *S3BufferClient) Get_Key() string
  • func (s *S3BufferClient) Init()
  • func (s *S3BufferClient) ReadyToFlush() bool
  • func (s *S3BufferClient) Remove_Flush()
  • func (s *S3BufferClient) Reset_Buffer()
• type Sequence

Functions

func FlushToS3

func FlushToS3(s *S3BufferClient, s3u s3manageriface.UploaderAPI) error

func Get_Key_Prefix

func Get_Key_Prefix() string

func PutRecordInBuffer

func PutRecordInBuffer(s *S3BufferClient, record []byte, seq Sequence, isOverflowData bool) (bool, error)

func UploadToS3

func UploadToS3(s3u s3manageriface.UploaderAPI, file io.Reader, bucket string, key string) error

Types

type Buffer

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

func (*Buffer) LockBuffer

func (b *Buffer) LockBuffer()

func (*Buffer) Read

func (b *Buffer) Read(p []byte) (n int, err error)

func (*Buffer) ReadFrom

func (b *Buffer) ReadFrom(r io.Reader) (n int64, err error)

func (*Buffer) Reset

func (b *Buffer) Reset()

func (*Buffer) String

func (b *Buffer) String() string

func (*Buffer) UnlockBuffer

func (b *Buffer) UnlockBuffer()

func (*Buffer) Write

func (b *Buffer) Write(p []byte) (n int, err error)

type S3BufferClient

type S3BufferClient struct {
	Env              string
	Shard_Id         string
	S3_Bucket        string
	Flush_Interval   int64
	Config           string
	MAX_BUFFER_SIZE  int64
	Compress_Level   int
	Buffer_Seq       Sequence
	Flushed_Seq_List []Sequence
	Buffer_Size      int64
	Buffer_Age       int64
	Last_Flush       int64
	Worker_Is_Alive  bool
	RecordBuffer     Buffer
	GzipBuffer       Buffer
	MetricSender     *metrics.SfxClient
	TmpFile          string
	Last_Flush_File  string
	Next_Flush_File  string
	LockFile         bool
	LockFileBuffer   [][]byte
	FilePointer      *os.File
}

func NewBufferClient

func NewBufferClient(Env string, Shard_Id string, S3_Bucket string, Region string, Flush_Interval int64, MAX_BUFFER_SIZE int64, Config string, metricSender *metrics.SfxClient) (*S3BufferClient, error)

func (*S3BufferClient) Background_Worker_Is_Dead

func (s *S3BufferClient) Background_Worker_Is_Dead() bool

func (*S3BufferClient) Begin_Background_Worker

func (s *S3BufferClient) Begin_Background_Worker(s3u s3manageriface.UploaderAPI)

func (*S3BufferClient) Get_Key

func (s *S3BufferClient) Get_Key() string

func (*S3BufferClient) Init

func (s *S3BufferClient) Init()

func (*S3BufferClient) ReadyToFlush

func (s *S3BufferClient) ReadyToFlush() bool

func (*S3BufferClient) Remove_Flush

func (s *S3BufferClient) Remove_Flush()

func (*S3BufferClient) Reset_Buffer

func (s *S3BufferClient) Reset_Buffer()

type Sequence

type Sequence struct {
	Sequence     int
	Sub_Sequence int
}