dispatcher

Documentation

Index

• Constants
• Variables
• func ActiveAgentCount() int
• func CheckAgents() error
• func FlushReports(maxErrDays, maxEmptyHours int)
• func HandleDeviceCreate(w http.ResponseWriter, r *http.Request)
• func HandleDeviceDelete(w http.ResponseWriter, r *http.Request)
• func HandleDeviceList(w http.ResponseWriter, r *http.Request)
• func HandleDeviceUpdate(w http.ResponseWriter, r *http.Request)
• func HandleDeviceUpsert(w http.ResponseWriter, r *http.Request)
• func HandleReport(w http.ResponseWriter, r *http.Request)
• func InitDB(dsn string) error
• func LoadAgents() error
• func PingHosts() ([]model.PingHost, error)
• func ReleaseDB()
• func RequestFromDB(devID int) (model.SnmpRequest, error)
• func RequestWithLock(id int) (model.SnmpRequest, error)
• func SendPingRequests(ctx context.Context)
• func SendPollingJobs(ctx context.Context)
• func SendRequest(ctx context.Context, req model.SnmpRequest, agent Agent) (stCode int, load float64, err error)
• func SnmpJobs() ([]int, error)
• func UnlockDevices()
• type Agent
  • func AgentsForDevice(devID int) []*Agent
  • func (a Agent) Check() (bool, float64)
  • func (a Agent) String() string
• type Agents
• type ByLoad
  • func (a ByLoad) Len() int
  • func (a ByLoad) Less(i, j int) bool
  • func (a ByLoad) Swap(i, j int)

Constants

const (
	// DeviceListURI is the api endpoint for device listing
	DeviceListURI = "/d/list"

	// DeviceCreateURI is the api endpoint for creating new device
	DeviceCreateURI = "/d/create"

	// DeviceUpdateURI is the api endpoint for updating a device
	DeviceUpdateURI = "/d/update"

	// DeviceUpsertURI is the api endpoint for `upserting` a device
	DeviceUpsertURI = "/d/upsert"

	// DeviceDeleteURI is the api endpoint for deleting a device
	DeviceDeleteURI = "/d/delete"
)

Variables

var (
	// MaxLoadDelta is the maximum average load difference allowed between agents
	// before moving a device to the least loaded agent.
	MaxLoadDelta float64 = 0.1

	// LoadAvgWindow is the window for agent snmp load average calculation.
	LoadAvgWindow time.Duration
)

var (
	// LocalIP is the local IP address used for API and report web server.
	// Defaults to the first non localhost address.
	LocalIP = getLocalIP()

	// Port is the web server port
	Port = 8080

	// HTTPTimeout is the timeout in seconds for posting poll requests
	HTTPTimeout = 3
)

var PingBatchCount int

PingBatchCount is the number of hosts per ping request, set at startup.

Functions

func ActiveAgentCount

func ActiveAgentCount() int

ActiveAgentCount returns the number of current active agents.

func CheckAgents

func CheckAgents() error

CheckAgents sends a keepalive to each agent and updates its status & current load.

func FlushReports

func FlushReports(maxErrDays, maxEmptyHours int)

FlushReports removes old report entries.

func HandleDeviceCreate

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

HandleDeviceCreate implements the CRUD create handler

func HandleDeviceDelete

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

HandleDeviceDelete implements the CRUD delete handler. The id of the device to delete must be given in `id` param to the POST request.

func HandleDeviceList

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

HandleDeviceList implements the CRUD list handler. When `id` parameter is given to the GET request, returns a json body with the device with this id. Otherwise, returns a json array with all devices ordered by id.

func HandleDeviceUpdate

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

HandleDeviceUpdate implements the CRUD update handler. All device required fields must be defined in the json as for the insert request.

func HandleDeviceUpsert

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

HandleDeviceUpsert implements the CRUD upsert handler. All device required fields must be defined in the json as for the insert request.

func HandleReport

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

HandleReport saves the polling report to db and unlocks the device.

func InitDB

func InitDB(dsn string) error

InitDB initializes db connection and prepares the db statements

func LoadAgents

func LoadAgents() error

LoadAgents retrieves agent list from db and updates the current agent list (removes deleted, adds new). Note: key for comparision is agent name (host:port).

func PingHosts

func PingHosts() ([]model.PingHost, error)

PingHosts retrieves alls hosts to be pinged.

func ReleaseDB

func ReleaseDB()

ReleaseDB closes the db connection.

func RequestFromDB

func RequestFromDB(devID int) (model.SnmpRequest, error)

RequestFromDB returns the request with the given device id from db.

func RequestWithLock

func RequestWithLock(id int) (model.SnmpRequest, error)

RequestWithLock builds a model.Request from the given device id and locks the device if there is no error.

func SendPingRequests

func SendPingRequests(ctx context.Context)

SendPingRequests sends current ping requests to agents with load-balancing and agent stickyness

func SendPollingJobs

func SendPollingJobs(ctx context.Context)

SendPollingJobs retrieves all available snmp polling jobs and sends each of them sequentially to the next available agent. If the request cannot be sent to an agent, we try the next one until the last. If no agent accepts the job, it is discarded.

func SendRequest

func SendRequest(ctx context.Context, req model.SnmpRequest, agent Agent) (stCode int, load float64, err error)

SendRequest sends the given request to the given agent. Returns the http status code, the agent's current load and an error if unsuccessful.

func SnmpJobs

func SnmpJobs() ([]int, error)

SnmpJobs returns a list of pollable device ids. A device is pollable if there is no ongoing polling job and was last polled past its polling frequency.

func UnlockDevices

func UnlockDevices()

UnlockDevices retrieves all ongoing requests from all active agents and unlocks all devices without any polling job and whose last job is past its global polling frequency. Is called periodically on a separate goroutine.

Types

type Agent

type Agent struct {
	// ID is the agent id
	ID int `db:"id"`

	// Host is the agent web server IP address
	Host string `db:"ip_address"`

	// Port is the agent web server listen port
	Port int `db:"port"`

	// Alive indicates wether this agent responds to keep-alives
	Alive bool `db:"is_alive"`
	// contains filtered or unexported fields
}

Agent represents an snmp agent

func AgentsForDevice

func AgentsForDevice(devID int) []*Agent

AgentsForDevice return a list of agents to which send a polling request by order of priority. We try to be sticky as much as possible but with balanced load: - the current list of active agents is sorted by load - if the device is not in jobDistrib map, this list is returned as is. - if the device is in jobDistrib map and its associated agent is active,

• if the load difference between the associated agent and the least loaded agent is under the MaxLoadDelta, we stick to this agent: a modified load-sorted list is returned where this agent is moved at the first position.
• if the load difference exceeds MaxLoadDelta, we rebalance the load: the load sorted list is returned.

func (Agent) Check

func (a Agent) Check() (bool, float64)

Check pings an agent and returns its active status and ongoing polls count. The check is a http query to the agents checkURL which returns a status 200 OK and the current load in body when it is healthy.

func (Agent) String

func (a Agent) String() string

String implements the stringer interface for the Agent type.

type Agents

type Agents map[string]*Agent

Agents is a map of Agent pointers with agent name as key.

type ByLoad

type ByLoad []*Agent

ByLoad is an Agent slice implementing Sort interface for sorting by average load.

func (ByLoad) Len

func (a ByLoad) Len() int

func (ByLoad) Less

func (a ByLoad) Less(i, j int) bool

func (ByLoad) Swap

func (a ByLoad) Swap(i, j int)