combainer

Documentation

Index

• Variables
• func Dashboard(w http.ResponseWriter, r *http.Request)
• func GenerateAndRegisterSerfResolver(lookup func() []serf.Member)
• func GetRouter(context ServerContext) http.Handler
• func Launch(s ServerContext, w http.ResponseWriter, r *http.Request)
• func NewSerfResolverBuilder(lookup func() []serf.Member) resolver.Builder
• func ParsingConfigs(s ServerContext, w http.ResponseWriter, r *http.Request)
• func ReadParsingConfig(s ServerContext, w http.ResponseWriter, r *http.Request)
• func Tasks(s ServerContext, w http.ResponseWriter, r *http.Request)
• type Client
  • func NewClient(opt ...func(*Client) error) (*Client, error)
  • func (cs *Client) AddFailedAggregate()
  • func (cs *Client) AddFailedParsing()
  • func (cs *Client) AddSuccessAggregate()
  • func (cs *Client) AddSuccessParsing()
  • func (cl *Client) Close() error
  • func (cs *Client) CopyStats(to *clientStats)
  • func (cl *Client) Dispatch(iteration uint64, parsingConfigName string, shouldWait bool) (string, error)
  • func (cs *Client) GetStats() *StatInfo
• type Cluster
  • func NewCluster(cfg repository.ClusterConfig) (*Cluster, error)
  • func (c *Cluster) AliveMembers() []serf.Member
  • func (c *Cluster) EventHandler()
  • func (c *Cluster) Hosts() []string
  • func (c *Cluster) IsLeader() bool
  • func (c *Cluster) Peers() ([]string, error)
  • func (c *Cluster) Run()
  • func (c *Cluster) Shutdown()
• type CombaineServer
  • func New(config CombaineServerConfig) (*CombaineServer, error)
  • func (c *CombaineServer) GetHosts() []string
  • func (c *CombaineServer) Serve() error
• type CombaineServerConfig
• type FSM
  • func (c *FSM) Apply(l *raft.Log) interface{}
  • func (c *FSM) Restore(rc io.ReadCloser) error
  • func (c *FSM) Snapshot() (raft.FSMSnapshot, error)
• type FSMCommand
• type FSMSnapshot
  • func (s *FSMSnapshot) Persist(sink raft.SnapshotSink) error
  • func (s *FSMSnapshot) Release()
• type FSMStore
  • func NewFSMStore() *FSMStore
  • func (s *FSMStore) DistributionStatistic() [][2]string
  • func (s *FSMStore) Dump() map[string][]string
  • func (s *FSMStore) List(host string) []string
  • func (s *FSMStore) Put(host, config string) chan struct{}
  • func (s *FSMStore) Remove(host, config string)
  • func (s *FSMStore) Replace(newStore map[string]map[string]chan struct{})
• type Observer
  • func (o *Observer) GetClientsStats() map[string]*StatInfo
  • func (o *Observer) RegisterClient(cl *Client, config string)
  • func (o *Observer) UnregisterClient(id uint64, config string)
• type OpenFiles
• type Resolver
  • func (r *Resolver) Close()
  • func (r *Resolver) ResolveNow(opt resolver.ResolveNowOption)
• type ServerContext
• type StatInfo

Variables

var GlobalObserver = Observer{
	// contains filtered or unexported fields
}

GlobalObserver is storage for client registations

Functions

func Dashboard

func Dashboard(w http.ResponseWriter, r *http.Request)

Dashboard handle http request abount internal statistics

func GenerateAndRegisterSerfResolver

func GenerateAndRegisterSerfResolver(lookup func() []serf.Member)

GenerateAndRegisterSerfResolver generates and registers Serf Resolver.

func GetRouter

func GetRouter(context ServerContext) http.Handler

GetRouter return mux root router

func Launch

func Launch(s ServerContext, w http.ResponseWriter, r *http.Request)

Launch run full iteration for config

func NewSerfResolverBuilder

func NewSerfResolverBuilder(lookup func() []serf.Member) resolver.Builder

NewSerfResolverBuilder creates a new serf resolver builder

func ParsingConfigs

func ParsingConfigs(s ServerContext, w http.ResponseWriter, r *http.Request)

ParsingConfigs list parsing configs names

func ReadParsingConfig

func ReadParsingConfig(s ServerContext, w http.ResponseWriter, r *http.Request)

ReadParsingConfig return parsing config content before return UpdateByCombainerConfig update config

func Tasks

func Tasks(s ServerContext, w http.ResponseWriter, r *http.Request)

Tasks return information about parsing tasks that should be performed by config

Types

type Client

type Client struct {
	ID uint64
	// contains filtered or unexported fields
}

Client is a distributor of tasks across the computation grid

func NewClient

func NewClient(opt ...func(*Client) error) (*Client, error)

NewClient returns new client

func (*Client) AddFailedAggregate

func (cs *Client) AddFailedAggregate()

func (*Client) AddFailedParsing

func (cs *Client) AddFailedParsing()

func (*Client) AddSuccessAggregate

func (cs *Client) AddSuccessAggregate()

func (*Client) AddSuccessParsing

func (cs *Client) AddSuccessParsing()

func (*Client) Close

func (cl *Client) Close() error

Close relases grpc.ClientConn

func (*Client) CopyStats

func (cs *Client) CopyStats(to *clientStats)

func (*Client) Dispatch

func (cl *Client) Dispatch(iteration uint64, parsingConfigName string, shouldWait bool) (string, error)

Dispatch does one iteration of tasks dispatching

func (*Client) GetStats

func (cs *Client) GetStats() *StatInfo

type Cluster

type Cluster struct {
	Name string
	// contains filtered or unexported fields
}

Cluster is wrapper for access cluster members

func NewCluster

func NewCluster(cfg repository.ClusterConfig) (*Cluster, error)

NewCluster create and initialize Cluster instance

func (*Cluster) AliveMembers

func (c *Cluster) AliveMembers() []serf.Member

AliveMembers return alive serf members

func (*Cluster) EventHandler

func (c *Cluster) EventHandler()

EventHandler is used to handle events from the serf cluster

func (*Cluster) Hosts

func (c *Cluster) Hosts() []string

Hosts return names of alive serf members

func (*Cluster) IsLeader

func (c *Cluster) IsLeader() bool

IsLeader checks if this server is the cluster leader

func (*Cluster) Peers

func (c *Cluster) Peers() ([]string, error)

Peers is used to return alive raft peers.

func (*Cluster) Run

func (c *Cluster) Run()

Run is used to monitor if we acquire or lose our role as the leader in the Raft cluster. If we are leader then we distribute tasks over cluster

func (*Cluster) Shutdown

func (c *Cluster) Shutdown()

Shutdown try gracefully shutdown raft cluster

type CombaineServer

type CombaineServer struct {
	Configuration   CombaineServerConfig
	CombainerConfig repository.CombainerConfig
	// contains filtered or unexported fields
}

CombaineServer main combaine object

func New

func New(config CombaineServerConfig) (*CombaineServer, error)

New create new combainer server

func (*CombaineServer) GetHosts

func (c *CombaineServer) GetHosts() []string

GetHosts return alive cluster members

func (*CombaineServer) Serve

func (c *CombaineServer) Serve() error

Serve run main event loop

type CombaineServerConfig

type CombaineServerConfig struct {
	// period of the locks rechecking
	Period time.Duration
	// Addrto listen for incoming http REST API requests
	RestEndpoint string
	//
	Active bool
}

CombaineServerConfig contains config from main combaine conf

type FSM

type FSM Cluster

FSM is cluster state

func (*FSM) Apply

func (c *FSM) Apply(l *raft.Log) interface{}

Apply command received over raft

func (*FSM) Restore

func (c *FSM) Restore(rc io.ReadCloser) error

Restore FSM from snapshot

func (*FSM) Snapshot

func (c *FSM) Snapshot() (raft.FSMSnapshot, error)

Snapshot create FSM snapshot

type FSMCommand

type FSMCommand struct {
	Type   string `json:"type"`
	Host   string `json:"host"`
	Config string `json:"config"`
}

FSMCommand contains cluster storage operation with data

type FSMSnapshot

type FSMSnapshot struct {
	Data []byte
}

FSMSnapshot ...

func (*FSMSnapshot) Persist

func (s *FSMSnapshot) Persist(sink raft.SnapshotSink) error

Persist ...

func (*FSMSnapshot) Release

func (s *FSMSnapshot) Release()

Release ...

type FSMStore

type FSMStore struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

FSMStore contains dispached congis

func NewFSMStore

func NewFSMStore() *FSMStore

NewFSMStore create new FSM storage

func (*FSMStore) DistributionStatistic

func (s *FSMStore) DistributionStatistic() [][2]string

DistributionStatistic dump number of configs assigned to hosts

func (*FSMStore) Dump

func (s *FSMStore) Dump() map[string][]string

Dump ...

func (*FSMStore) List

func (s *FSMStore) List(host string) []string

List return configs assigned to host

func (*FSMStore) Put

func (s *FSMStore) Put(host, config string) chan struct{}

Put assign new config to host

func (*FSMStore) Remove

func (s *FSMStore) Remove(host, config string)

Remove remove config from host's store

func (*FSMStore) Replace

func (s *FSMStore) Replace(newStore map[string]map[string]chan struct{})

Replace store for testing

type Observer

type Observer struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Observer object with registered clients

func (*Observer) GetClientsStats

func (o *Observer) GetClientsStats() map[string]*StatInfo

GetClientsStats return map with client stats

func (*Observer) RegisterClient

func (o *Observer) RegisterClient(cl *Client, config string)

RegisterClient register client in Observer ReRegister client is UnregisterClient for previously registered client, but all stats are copied

func (*Observer) UnregisterClient

func (o *Observer) UnregisterClient(id uint64, config string)

UnregisterClient unregister client in Observer Deregister only a yourself by checking id

type OpenFiles

type OpenFiles struct {
	Open  uint64
	Limit syscall.Rlimit
}

OpenFiles contains info abound fd usage

type Resolver

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

Resolver is Serf members resolver

func (*Resolver) Close

func (r *Resolver) Close()

Close is a noop for Resolver.

func (*Resolver) ResolveNow

func (r *Resolver) ResolveNow(opt resolver.ResolveNowOption)

ResolveNow invoke an immediate resolution of the target that this serfResolver watches.

type ServerContext

type ServerContext interface {
	GetHosts() []string
}

ServerContext contains server context with repository

type StatInfo

type StatInfo struct {
	ParsingSuccess   int64
	ParsingFailed    int64
	ParsingTotal     int64
	AggregateSuccess int64
	AggregateFailed  int64
	AggregateTotal   int64
	Heartbeated      int64
}

StatInfo contains stats about main operations (aggregating and parsing)