change.go

package pnd

import (
	"errors"
	"github.com/google/uuid"
	"github.com/openconfig/ygot/ygot"
	log "github.com/sirupsen/logrus"
	"golang.org/x/net/context"
	"os"
	"sync"
	"time"
)

var changeTimeout time.Duration

func init() {
	var err error
	e := os.Getenv("GOSDN_CHANGE_TIMEOUT")
	if e != "" {
		changeTimeout, err = time.ParseDuration(e)
		if err != nil {
			log.Fatal(err)
		}
	} else {
		changeTimeout, err = time.ParseDuration("10m")
		if err != nil {
			log.Fatal(err)
		}
	}
	log.Debugf("change timeout set to %v", changeTimeout)
}

// NewChange takes a Device UUID, a pair GoStructs (current and intended state)
// a callback function and a channel for errors and returns a *Change
// The callback function is used by the Commit() and Confirm() functions. It
// must define how the change is carried out.
func NewChange(device uuid.UUID, currentState ygot.GoStruct, change ygot.GoStruct, callback func(ygot.GoStruct, ygot.GoStruct) error, errChan chan error) *Change {
	return &Change{
		cuid:          uuid.New(),
		duid:          device,
		timestamp:     time.Now(),
		previousState: currentState,
		intendedState: change,
		committed:     false,
		confirmed:     false,
		callback:      callback,
		errChan:       errChan,
		Done:          make(chan int),
	}
}

// Change is an intended change to an OND. It is unique and immutable.
// It has a cuid, a timestamp, and holds both the previous and the new
// state. It keeps track if the state is committed and confirmed. A callback
// exists to acess the proper transport for the changed OND
type Change struct {
	cuid          uuid.UUID
	duid          uuid.UUID
	timestamp     time.Time
	previousState ygot.GoStruct
	intendedState ygot.GoStruct
	committed     bool
	confirmed     bool
	inconsistent  bool
	callback      func(ygot.GoStruct, ygot.GoStruct) error
	lock          sync.RWMutex
	cancelFunc    context.CancelFunc
	errChan       chan error
	// TODO: Move nucleus.pndImplementation and Change to same package and unexport
	Done chan int
}

// ID returns the Change's UUID
func (c *Change) ID() uuid.UUID {
	return c.cuid
}

// Commit pushes the cange to the OND using the callback() function
// and starts the timeout-timer for the Change. If the timer expires
// the change is rolled back.
func (c *Change) Commit() error {
	if err := c.callback(c.intendedState, c.previousState); err != nil {
		return err
	}
	c.committed = true
	log.WithFields(log.Fields{
		"change uuid": c.cuid,
		"device uuid": c.duid,
	}).Debug("change commited")
	ctx, cancel := context.WithCancel(context.Background())
	c.cancelFunc = cancel
	go c.rollbackHandler(ctx)
	return nil
}

func (c *Change) rollbackHandler(ctx context.Context) {
	select {
	case <-ctx.Done():
		return
	case <-time.Tick(changeTimeout):
		c.lock.RLock()
		defer c.lock.RUnlock()
		if !c.confirmed {
			c.errChan <- c.callback(c.previousState, c.intendedState)
			log.WithFields(log.Fields{
				"change uuid": c.cuid,
				"device uuid": c.duid,
			}).Info("change timed out")
		}
	}
}

// Confirm confirms a committed Change and stops the rollback timer.
func (c *Change) Confirm() error {
	c.lock.RLock()
	if !c.committed {
		defer c.lock.RUnlock()
		return errors.New("cannot confirm uncommitted change")
	}
	c.lock.RUnlock()
	c.lock.Lock()
	defer c.lock.Unlock()
	c.confirmed = true
	c.cancelFunc()
	close(c.errChan)
	c.Done <- 0
	close(c.Done)
	log.WithFields(log.Fields{
		"change uuid": c.cuid,
		"device uuid": c.duid,
	}).Info("change confirmed")
	return nil
}