s3spanstore

Aug 9, 2023

Changes in this version

type Reader

+ func (r *Reader) Close() error

Mar 1, 2022

Changes in this version

+ type DependenciesPrefetch struct

+ func NewDependenciesPrefetch(ctx context.Context, logger hclog.Logger, reader ReaderWithDependencies, ...) *DependenciesPrefetch

+ func (d *DependenciesPrefetch) Start()

+ func (d *DependenciesPrefetch) Stop()

+ type ReaderWithDependencies interface

+ GetDependencies func(ctx context.Context, endTs time.Time, lookback time.Duration) ([]model.DependencyLink, error)

Mar 1, 2022

Changes in this version

+ type DeduplicatableRow interface

+ DedupeKey func() string

type OperationRecord

+ func (w *OperationRecord) DedupeKey() string

Mar 1, 2022

Changes in this version

+ type DedupeParquetWriter struct

+ func NewDedupeParquetWriter(logger hclog.Logger, dedupeDuration time.Duration, dedupeCacheSize int, ...) (*DedupeParquetWriter, error)

+ func (w *DedupeParquetWriter) Close() error

+ func (w *DedupeParquetWriter) Write(ctx context.Context, rowTime time.Time, row interface{}) error

+ type IParquetWriter interface

+ Close func() error

+ Write func(ctx context.Context, time time.Time, row interface{}) error

+ type OperationRecord struct

+ OperationName string

+ ServiceName string

+ SpanKind string

+ func NewOperationRecordFromSpan(span *model.Span) (*OperationRecord, error)

+ type ParquetWriter struct

+ func NewParquetWriter(ctx context.Context, logger hclog.Logger, svc S3API, ...) (*ParquetWriter, error)

+ func (w *ParquetWriter) Close() error

+ func (w *ParquetWriter) Write(ctx context.Context, time time.Time, row interface{}) error

Feb 21, 2022

Changes in this version

+ type AthenaAPI interface

+ BatchGetQueryExecution func(ctx context.Context, params *athena.BatchGetQueryExecutionInput, ...) (*athena.BatchGetQueryExecutionOutput, error)

+ GetQueryExecution func(ctx context.Context, params *athena.GetQueryExecutionInput, ...) (*athena.GetQueryExecutionOutput, error)

+ GetQueryResults func(ctx context.Context, params *athena.GetQueryResultsInput, ...) (*athena.GetQueryResultsOutput, error)

+ ListQueryExecutions func(ctx context.Context, params *athena.ListQueryExecutionsInput, ...) (*athena.ListQueryExecutionsOutput, error)

+ StartQueryExecution func(ctx context.Context, params *athena.StartQueryExecutionInput, ...) (*athena.StartQueryExecutionOutput, error)

+ StopQueryExecution func(ctx context.Context, params *athena.StopQueryExecutionInput, ...) (*athena.StopQueryExecutionOutput, error)

+ type AthenaQueryCache struct

+ func NewAthenaQueryCache(logger hclog.Logger, svc AthenaAPI, workGroup string) *AthenaQueryCache

+ func (c *AthenaQueryCache) Lookup(ctx context.Context, key string, ttl time.Duration) (*types.QueryExecution, error)

Feb 3, 2022

Changes in this version

+ func S3PartitionKey(t time.Time) string

+ type ParquetRef struct

type Reader

+ func (r *Reader) DefaultMaxTime() time.Time

+ func (r *Reader) DefaultMinTime() time.Time

+ func (r *Reader) GetDependencies(ctx context.Context, endTs time.Time, lookback time.Duration) ([]model.DependencyLink, error)

Feb 2, 2022

Changes in this version

+ const ATHENA_TIMEFORMAT

+ const PARTION_FORMAT

+ func DecodeSpanPayload(payload string) (*model.Span, error)

+ func EmptyBucket(ctx context.Context, svc S3API, bucketName string) error

type S3API

+ DeleteObject func(ctx context.Context, params *s3.DeleteObjectInput, optFns ...func(*s3.Options)) (*s3.DeleteObjectOutput, error)

+ ListObjectsV2 func(ctx context.Context, params *s3.ListObjectsV2Input, ...) (*s3.ListObjectsV2Output, error)

Jan 30, 2022

Changes in this version

+ func S3ParquetKey(prefix, suffix string, t time.Time) string

Jan 28, 2022

Changes in this version

+ const PARQUET_CONCURRENCY

+ var PARQUET_ROTATION_INTERVAL = 60 * time.Second

+ func RandStringBytes(n int) string

+ type S3API interface

+ AbortMultipartUpload func(context.Context, *s3.AbortMultipartUploadInput, ...func(*s3.Options)) (*s3.AbortMultipartUploadOutput, error)

+ CompleteMultipartUpload func(context.Context, *s3.CompleteMultipartUploadInput, ...func(*s3.Options)) (*s3.CompleteMultipartUploadOutput, error)

+ CreateMultipartUpload func(context.Context, *s3.CreateMultipartUploadInput, ...func(*s3.Options)) (*s3.CreateMultipartUploadOutput, error)

+ GetObject func(context.Context, *s3.GetObjectInput, ...func(*s3.Options)) (*s3.GetObjectOutput, error)

+ HeadObject func(ctx context.Context, params *s3.HeadObjectInput, optFns ...func(*s3.Options)) (*s3.HeadObjectOutput, error)

+ PutObject func(context.Context, *s3.PutObjectInput, ...func(*s3.Options)) (*s3.PutObjectOutput, error)

+ UploadPart func(context.Context, *s3.UploadPartInput, ...func(*s3.Options)) (*s3.UploadPartOutput, error)

Jan 26, 2022

Changes in this version

+ func EncodeSpanPayload(span *model.Span) (string, error)

Jan 25, 2022

Changes in this version

+ const MAX_BATCH_BYTE_SIZE

+ const MAX_BATCH_RECORDS

+ var MIN_FLUSH_INTERVAL = 10 * time.Second

type KinesisAPI

+ PutRecordBatch func(ctx context.Context, params *firehose.PutRecordBatchInput, ...) (*firehose.PutRecordBatchOutput, error)

type Writer

+ func (w *Writer) Close() error

Jan 24, 2022

Changes in this version

type SpanRecord

+ References []*SpanRecordReferences

+ type SpanRecordReferences struct

+ RefType int64

+ SpanID string

+ TraceID string

+ func NewSpanRecordReferencesFromSpanReferences(span *model.Span) []*SpanRecordReferences

Jan 23, 2022

Changes in this version

+ type KinesisAPI interface

+ PutRecord func(ctx context.Context, params *firehose.PutRecordInput, ...) (*firehose.PutRecordOutput, error)

+ type Reader struct

+ func NewReader(logger hclog.Logger, svc *athena.Client, cfg config.Athena) (*Reader, error)

+ func (s *Reader) FindTraceIDs(ctx context.Context, query *spanstore.TraceQueryParameters) ([]model.TraceID, error)

+ func (s *Reader) FindTraces(ctx context.Context, query *spanstore.TraceQueryParameters) ([]*model.Trace, error)

+ func (s *Reader) GetOperations(ctx context.Context, query spanstore.OperationQueryParameters) ([]spanstore.Operation, error)

+ func (s *Reader) GetServices(ctx context.Context) ([]string, error)

+ func (s *Reader) GetTrace(ctx context.Context, traceID model.TraceID) (*model.Trace, error)

+ type SpanRecord struct

+ Duration int64

+ OperationName string

+ ServiceName string

+ SpanID string

+ SpanKind string

+ SpanPayload string

+ StartTime int64

+ Tags map[string]string

+ TraceID string

+ func NewSpanRecordFromSpan(span *model.Span) (*SpanRecord, error)

+ type Writer struct

+ func NewWriter(logger hclog.Logger, svc KinesisAPI, firehoseConfig config.Kinesis) (*Writer, error)

+ func (s *Writer) WriteSpan(ctx context.Context, span *model.Span) error