Files
mgmt/resources/timer.go
James Shubin 2e718c0e9d resources: Improve notification system and notify refreshes
Resources can send "refresh" notifications along edges. These messages
are sent whenever the upstream (initiating vertex) changes state. When
the changed state propagates downstream, it will be paired with a
refresh flag which can be queried in the CheckApply method of that
resource.

Future work will include a stateful refresh tracking mechanism so that
if a refresh event is generated and not consumed, it will be saved
across an interrupt (shutdown) or a crash so that it can be re-applied
on the subsequent run. This is important because the unapplied refresh
is a form of hysteresis which needs to be tracked and remembered or we
won't be able to determine that the state is wrong!

Still to do:
* Update the autogrouping code to handle the edge notify properties!
* Actually finish the stateful bool code
2016-12-03 01:35:31 -05:00

184 lines
4.7 KiB
Go

// Mgmt
// Copyright (C) 2013-2016+ James Shubin and the project contributors
// Written by James Shubin <james@shubin.ca> and the project contributors
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package resources
import (
"encoding/gob"
"log"
"time"
"github.com/purpleidea/mgmt/event"
)
func init() {
gob.Register(&TimerRes{})
}
// TimerRes is a timer resource for time based events.
type TimerRes struct {
BaseRes `yaml:",inline"`
Interval int `yaml:"interval"` // Interval : Interval between runs
ticker *time.Ticker
}
// TimerUID is the UID struct for TimerRes.
type TimerUID struct {
BaseUID
name string
}
// NewTimerRes is a constructor for this resource. It also calls Init() for you.
func NewTimerRes(name string, interval int) (*TimerRes, error) {
obj := &TimerRes{
BaseRes: BaseRes{
Name: name,
},
Interval: interval,
}
return obj, obj.Init()
}
// Init runs some startup code for this resource.
func (obj *TimerRes) Init() error {
obj.BaseRes.kind = "Timer"
return obj.BaseRes.Init() // call base init, b/c we're overrriding
}
// Validate the params that are passed to TimerRes
// Currently we are getting only an interval in seconds
// which gets validated by go compiler
func (obj *TimerRes) Validate() error {
return nil
}
// newTicker creates a new ticker
func (obj *TimerRes) newTicker() *time.Ticker {
return time.NewTicker(time.Duration(obj.Interval) * time.Second)
}
// Watch is the primary listener for this resource and it outputs events.
func (obj *TimerRes) Watch(processChan chan event.Event) error {
if obj.IsWatching() {
return nil
}
obj.SetWatching(true)
defer obj.SetWatching(false)
cuid := obj.converger.Register()
defer cuid.Unregister()
var startup bool
Startup := func(block bool) <-chan time.Time {
if block {
return nil // blocks forever
//return make(chan time.Time) // blocks forever
}
return time.After(time.Duration(500) * time.Millisecond) // 1/2 the resolution of converged timeout
}
// create a time.Ticker for the given interval
obj.ticker = obj.newTicker()
defer obj.ticker.Stop()
var send = false
for {
obj.SetState(ResStateWatching)
select {
case <-obj.ticker.C: // received the timer event
send = true
log.Printf("%s[%s]: received tick", obj.Kind(), obj.GetName())
case event := <-obj.Events():
cuid.SetConverged(false)
if exit, _ := obj.ReadEvent(&event); exit {
return nil
}
case <-cuid.ConvergedTimer():
cuid.SetConverged(true)
continue
case <-Startup(startup):
cuid.SetConverged(false)
send = true
}
if send {
startup = true // startup finished
send = false
if exit, err := obj.DoSend(processChan, "timer ticked"); exit || err != nil {
return err // we exit or bubble up a NACK...
}
}
}
}
// CheckApply method for Timer resource. Triggers a timer reset on notify.
func (obj *TimerRes) CheckApply(apply bool) (bool, error) {
// because there are no checks to run, this resource has a less
// traditional pattern than what is seen in most resources...
if !obj.Refresh() { // this works for apply || !apply
return true, nil // state is always okay if no refresh to do
} else if !apply { // we had a refresh to do
return false, nil // therefore state is wrong
}
// reset the timer since apply && refresh
obj.ticker.Stop()
obj.ticker = obj.newTicker()
return false, nil
}
// GetUIDs includes all params to make a unique identification of this object.
// Most resources only return one, although some resources can return multiple.
func (obj *TimerRes) GetUIDs() []ResUID {
x := &TimerUID{
BaseUID: BaseUID{
name: obj.GetName(),
kind: obj.Kind(),
},
name: obj.Name,
}
return []ResUID{x}
}
// AutoEdges returns the AutoEdge interface. In this case no autoedges are used.
func (obj *TimerRes) AutoEdges() AutoEdge {
return nil
}
// Compare two resources and return if they are equivalent.
func (obj *TimerRes) Compare(res Res) bool {
switch res.(type) {
case *TimerRes:
res := res.(*TimerRes)
if !obj.BaseRes.Compare(res) {
return false
}
if obj.Name != res.Name {
return false
}
if obj.Interval != res.Interval {
return false
}
default:
return false
}
return true
}