log

README

##Log validation service

Usage:

1. register log listener, to dynamically detect any log changes (log shrinking/rotation is supported), any new log records are queued to be validated.
2. run log validation. Log validation verifies actual and expected log records, shifting record from actual logs pending queue.
3. reset - optionally reset log queues, to discard pending validation logs.

Supported actions:

Service Id	Action	Description	Request	Response
validator/log	listen	start listening for log file changes on specified location	ListenRequest	ListenResponse
validator/log	reset	discard logs detected by listener	ResetRequest	ResetResponse
validator/log	assert	perform validation on provided expected log records against actual log file records.	AssertRequest	AssertResponse

Validation strategies:

A log validation verifies produced by a logger with a user provides a desired log records in the asset request.Expect[logTypeIndex].Records Any arbitrary data structure can represent records.

Once a log/validator listener detects data produce by a logger, it places it to the pending validation queue, then later when assert request takes place, validator takes (and removes) records from pending validation queue to match and validate with expected records.

This process may use either position or index based matching method. In the first strategy, a matcher takes the first record from the pending validation queue (FIFO) for each expected record. The latter strategy requires an indexing expression (provided in listen request IndexRegExpr i.e. "UUID":"([^"]+)" ) which is used for both indexing pending logs and desired logs. If the validator is unable to match record with indexing expression, it falls back to the position based one.

Validator also supports data transformation on the fly just before validation with UDF

Actual validation is delegated to assertly

Examples

Standalone testing workflow example:**

endly -r=run

@run.yaml

init:
  logLocation: /tmp/logs
  target:
    url:  ssh://127.0.0.1/
    credentials: ${env.HOME}/.secret/localhost.json
defaults:
  target: $target
pipeline:
  init:
    action: exec:run
    commands:
      - mkdir -p $logLocation
      - "> ${logLocation}/myevents.log"
      - echo '{"EventID":111, "EventType":"event1", "X":11111111}' >> ${logLocation}/myevents.log
      - echo '{"EventID":222, "EventType":"event2", "X":11141111}' >> ${logLocation}/myevents.log
      - echo '{"EventID":333, "EventType":"event1","X":22222222}' >>  ${logLocation}/myevents.log
  listen:
    action: validator/log:listen
    frequencyMs: 500
    source:
      URL: $logLocation
    types:
      - format: json
        inclusion: event1
        mask: '*.log'
        name: event1
  validate:
    action: validator/log:assert
    logTypes:
      - event1
    description: E-logger event log validation
    expect:
      - type: event1
        records:
         - EventID: 111
           X: 11111111
         - EventID: 333
           X: 22222222
    logWaitRetryCount: 2
    logWaitTimeMs: 5000

with request delegation:

endly -r=test

@test.yaml

init:
  logLocation: /tmp/logs
  target:
    url:  ssh://127.0.0.1/
    credentials: ${env.HOME}/.secret/localhost.json
defaults:
  target: $target
pipeline:
  init:
    action: exec:run
    request: '@exec.yaml'
  listen:
    action: validator/log:listen
    request: '@listen.yaml'
  validate:
    action: validator/log:assert
    request: '@validate.json'

@exec.yaml

commands:
  - mkdir -p $logLocation
  - "> ${logLocation}/myevents.log"
  - echo '{"EventID":111, "EventType":"event1", "X":11111111}' >> ${logLocation}/myevents.log
  - echo '{"EventID":222, "EventType":"event2", "X":11141111}' >> ${logLocation}/myevents.log
  - echo '{"EventID":333, "EventType":"event1","X":22222222}' >>  ${logLocation}/myevents.log

@listen.yaml

frequencyMs: 500
source:
  URL: $logLocation
types:
  - format: json
    inclusion: event1
    mask: '*.log'
    name: event1

@validate.json

{
  "Expect": [
    {
      "type": "event1",
      "records": [
        {
          "EventID": 111,
          "X": 11111111
        },
        {
          "EventID": 333,
          "X": 22222222
        }
      ]
    }
  ]
}

Workflow with csv UDF example

endly -r=csv

csv.yaml

init:
  i: 0
  j: 0
  logLocation: /tmp/logs
  target:
    url:  ssh://127.0.0.1/
    credentials: ${env.HOME}/.secret/localhost.json
defaults:
  target: $target
pipeline:
  init:
    make-dir:
      action: exec:run
      commands:
      - mkdir -p $logLocation
      - "> ${logLocation}/events.csv"
    register-udf:
      action: udf:register
      udfs:
        - id: UserCsvReader
          provider: CsvReader
          params:
            - id,type,timestamp,user
    listen:
      action: validator/log:listen
      frequencyMs: 500
      source:
        URL: $logLocation
      types:
        - format: json
          inclusion: event1
          mask: '*.csv'
          name: event1
          UDF: UserCsvReader
          debug: true
  test:
    multiAction: true
    produce:
      async: true
      repeat: 6
      sleepTimeMs: 400
      action: exec:run
      commands:
        - echo '$i++,event1,${timestamp.now},user $j++' >> ${logLocation}/events.csv

    validate:
      action: validator/log:assert
      logTypes:
        - event1
      description: E-logger event log validation
      expect:
      - type: event1
        records:
          - id: 0
            user: user 0
          - id: 1
            user: user 1
          - id: 2
            user: user 2
          - id: 3
            user: user 3
          - id: 4
            user: user 4
          - id: 5
            user: user 5

      logWaitRetryCount: 10
      logWaitTimeMs: 2000

As part of workflow

See more

Documentation

Index

• Constants
• func New() endly.Service
• type AssertRequest
  • func (r *AssertRequest) Init() error
  • func (r *AssertRequest) Validate() error
• type AssertResponse
  • func (r *AssertResponse) Assertion() []*assertly.Validation
• type File
  • func (f *File) HasPendingLogs() bool
  • func (f *File) PushLogRecord(record *Record)
  • func (f *File) Reset(object storage.Object)
  • func (f *File) ShiftLogRecord() *Record
  • func (f *File) ShiftLogRecordByIndex(value string) *Record
• type IndexedRecord
• type ListenRequest
• type ListenResponse
• type ProcessingState
  • func (s *ProcessingState) Reset()
  • func (s *ProcessingState) Update(position, lineNumber int) (string, int)
• type Record
  • func (r *Record) AsMap() (map[string]interface{}, error)
• type ResetRequest
• type ResetResponse
• type Type
  • func (t *Type) GetIndexExpr() (*regexp.Regexp, error)
  • func (t *Type) UseIndex() bool
• type TypeMeta
  • func NewTypeMeta(source *url.Resource, logType *Type) *TypeMeta
  • func (m *TypeMeta) Iterator() toolbox.Iterator
• type TypedRecord
• type TypesMeta

Constants

const (
	//ServiceID represents log validator service id.
	ServiceID = "validator/log"
)

Functions

func New

func New() endly.Service

New creates a new log validator service.

Types

type AssertRequest

type AssertRequest struct {
	LogWaitTimeMs       int
	LogWaitRetryCount   int
	Description         string
	DescriptionTemplate string
	Expect              []*TypedRecord `required:"true" description:"expected log data"`
}

AssertRequest represents a log assert request

func (*AssertRequest) Init

func (r *AssertRequest) Init() error

Init converts yaml kv pairs to a map if applicable

func (*AssertRequest) Validate

func (r *AssertRequest) Validate() error

Validate check if request is valid

type AssertResponse

type AssertResponse struct {
	Validations []*assertly.Validation
}

AssertResponse represents a log assert response

func (*AssertResponse) Assertion

func (r *AssertResponse) Assertion() []*assertly.Validation

Assertion returns description with validation slice

type File

type File struct {
	URL     string
	Content string
	Name    string
	*Type
	ProcessingState *ProcessingState
	LastModified    time.Time
	Size            int
	Records         []*Record
	IndexedRecords  map[string]*Record
	Mutex           *sync.RWMutex
	// contains filtered or unexported fields
}

File represents a log file

func (*File) HasPendingLogs

func (f *File) HasPendingLogs() bool

HasPendingLogs returns true if file has pending validation records

func (*File) PushLogRecord

func (f *File) PushLogRecord(record *Record)

PushLogRecord appends provided log record to the records.

func (*File) Reset

func (f *File) Reset(object storage.Object)

Reset resets processing state

func (*File) ShiftLogRecord

func (f *File) ShiftLogRecord() *Record

ShiftLogRecord returns and remove the first log record if present

func (*File) ShiftLogRecordByIndex

func (f *File) ShiftLogRecordByIndex(value string) *Record

ShiftLogRecordByIndex returns and remove the first log record if present

type IndexedRecord

type IndexedRecord struct {
	*Record
	IndexValue string
}

IndexedRecord represents indexed log record

type ListenRequest

type ListenRequest struct {
	FrequencyMs int
	Source      *url.Resource `required:"true" description:"log location"`
	Types       []*Type       `required:"true" description:"log types"`
}

ListenRequest represents listen for a logs request.

type ListenResponse

type ListenResponse struct {
	Meta TypesMeta
}

ListenResponse represents a log validation listen response.

type ProcessingState

type ProcessingState struct {
	Line     int
	Position int
}

ProcessingState represents log processing state

func (*ProcessingState) Reset

func (s *ProcessingState) Reset()

Reset resets processing state

func (*ProcessingState) Update

func (s *ProcessingState) Update(position, lineNumber int) (string, int)

Update updates processed position and line number

type Record

type Record struct {
	URL    string
	Number int
	Line   string
}

Record represents a log record

func (*Record) AsMap

func (r *Record) AsMap() (map[string]interface{}, error)

AsMap returns log records as map

type ResetRequest

type ResetRequest struct {
	LogTypes []string `required:"true" description:"log types to reset"`
}

ResetRequest represents a log reset request

type ResetResponse

type ResetResponse struct {
	LogFiles []string
}

ResetResponse represents a log reset response

type Type

type Type struct {
	Name         string `required:"true" description:"log type name"`
	Format       string `description:"log format"`
	Mask         string `description:"expected log file mast"`
	Exclusion    string `description:"if specified, exclusion fragment can not match log record"`
	Inclusion    string `description:"if specified, inclusion fragment must match log record"`
	IndexRegExpr string `description:"provide expression for indexing log messages, in this case position based logging will not apply"` //provide expression for indexing log message, in this case position based logging will not apply

	UDF   string `description:"registered user defined function to transform content file before applying validation"`
	Debug bool   `description:"if set, every record appended to validation queue will be listed"`
	// contains filtered or unexported fields
}

Type represents a log type

func (*Type) GetIndexExpr

func (t *Type) GetIndexExpr() (*regexp.Regexp, error)

GetIndexExpr returns index expression.

func (*Type) UseIndex

func (t *Type) UseIndex() bool

UseIndex returns true if index can be used.

type TypeMeta

type TypeMeta struct {
	Source   *url.Resource
	LogType  *Type
	LogFiles map[string]*File
}

TypeMeta represents a log type meta

func NewTypeMeta

func NewTypeMeta(source *url.Resource, logType *Type) *TypeMeta

NewTypeMeta creates a nre log type meta.

func (*TypeMeta) Iterator

func (m *TypeMeta) Iterator() toolbox.Iterator

Iterator returns log record iterator

type TypedRecord

type TypedRecord struct {
	TagID   string `description:"neatly tag id for matching validation summary"`
	Type    string `required:"true" description:"log type register with listener"`
	Records []interface{}
}

TypedRecord represents an expected log record.

type TypesMeta

type TypesMeta map[string]*TypeMeta

TypesMeta represents log type meta details