resources: Implement Send -> Recv

This is a new design idea which I had. Whether it stays around or not is
up for debate. For now it's a rough POC.

The idea is that any resource can _produce_ data, and any resource can
_consume_ data. This is what we call send and recv. By linking the two
together, data can be passed directly between resources, which will
maximize code re-use, and allow for some interesting logical graphs.

For example, you might have an HTTP resource which puts its output in a
particular file. This avoids having to overload the HTTP resource with
all of the special behaviours of the File resource.

For our POC, I implemented a `password` resource which generates a
random string which can then be passed to a receiver such as a file. At
this point the password resource isn't recommended for sensitive
applications because it caches the password as plain text.

Still to do:
* Statically check all of the type matching before we run the graph
* Verify that our autogrouping works correctly around this feature
* Verify that appropriate edges exist between send->recv pairs
* Label the password as generated instead of storing the plain text
* Consider moving password logic from Init() to CheckApply()
* Consider combining multiple send values (list?) into a single receiver
* Consider intermediary transformation nodes for value combining
This commit is contained in:
James Shubin
2016-11-23 16:25:33 -05:00
parent 63c5e35e2b
commit 7f1c13a576
11 changed files with 669 additions and 16 deletions

View File

@@ -222,6 +222,7 @@ parameter with the [Noop](#Noop) resource.
* [Msg](#Msg): Send log messages.
* [Noop](#Noop): A simple resource that does nothing.
* [Nspawn](#Nspawn): Manage systemd-machined nspawn containers.
* [Password](#Password): Create random password strings.
* [Pkg](#Pkg): Manage system packages with PackageKit.
* [Svc](#Svc): Manage system systemd services.
* [Timer](#Timer): Manage system systemd services.
@@ -303,6 +304,10 @@ The noop resource does absolutely nothing. It does have some utility in testing
The nspawn resource is used to manage systemd-machined style containers.
###Password
The password resource can generate a random string to be used as a password.
###Pkg
The pkg resource is used to manage system packages. This resource works on many

192
examples/lib/libmgmt3.go Normal file
View File

@@ -0,0 +1,192 @@
// libmgmt example of send->recv
package main
import (
"fmt"
"log"
"os"
"os/signal"
"sync"
"syscall"
"time"
"github.com/purpleidea/mgmt/gapi"
mgmt "github.com/purpleidea/mgmt/mgmtmain"
"github.com/purpleidea/mgmt/pgraph"
"github.com/purpleidea/mgmt/resources"
)
// MyGAPI implements the main GAPI interface.
type MyGAPI struct {
Name string // graph name
Interval uint // refresh interval, 0 to never refresh
data gapi.Data
initialized bool
closeChan chan struct{}
wg sync.WaitGroup // sync group for tunnel go routines
}
// NewMyGAPI creates a new MyGAPI struct and calls Init().
func NewMyGAPI(data gapi.Data, name string, interval uint) (*MyGAPI, error) {
obj := &MyGAPI{
Name: name,
Interval: interval,
}
return obj, obj.Init(data)
}
// Init initializes the MyGAPI struct.
func (obj *MyGAPI) Init(data gapi.Data) error {
if obj.initialized {
return fmt.Errorf("Already initialized!")
}
if obj.Name == "" {
return fmt.Errorf("The graph name must be specified!")
}
obj.data = data // store for later
obj.closeChan = make(chan struct{})
obj.initialized = true
return nil
}
// Graph returns a current Graph.
func (obj *MyGAPI) Graph() (*pgraph.Graph, error) {
if !obj.initialized {
return nil, fmt.Errorf("libmgmt: MyGAPI is not initialized")
}
g := pgraph.NewGraph(obj.Name)
p1 := &resources.PasswordRes{
BaseRes: resources.BaseRes{
Name: "password1",
},
Length: 8, // generated string will have this many characters
}
v1 := pgraph.NewVertex(p1)
g.AddVertex(v1)
f1 := &resources.FileRes{
BaseRes: resources.BaseRes{
Name: "file1",
// send->recv!
Recv: map[string]resources.Send{
"Content": resources.Send{Res: p1, Key: "Password"},
},
},
Path: "/tmp/mgmt/f1",
//Content: p1.Password, // won't work
State: "present",
}
v2 := pgraph.NewVertex(f1)
g.AddVertex(v2)
g.AddEdge(v1, v2, pgraph.NewEdge("e1"))
//g, err := config.NewGraphFromConfig(obj.data.Hostname, obj.data.EmbdEtcd, obj.data.Noop)
return g, nil
}
// SwitchStream returns nil errors every time there could be a new graph.
func (obj *MyGAPI) SwitchStream() chan error {
if obj.data.NoWatch || obj.Interval <= 0 {
return nil
}
ch := make(chan error)
obj.wg.Add(1)
go func() {
defer obj.wg.Done()
defer close(ch) // this will run before the obj.wg.Done()
if !obj.initialized {
ch <- fmt.Errorf("libmgmt: MyGAPI is not initialized")
return
}
// arbitrarily change graph every interval seconds
ticker := time.NewTicker(time.Duration(obj.Interval) * time.Second)
defer ticker.Stop()
for {
select {
case <-ticker.C:
log.Printf("libmgmt: Generating new graph...")
ch <- nil // trigger a run
case <-obj.closeChan:
return
}
}
}()
return ch
}
// Close shuts down the MyGAPI.
func (obj *MyGAPI) Close() error {
if !obj.initialized {
return fmt.Errorf("libmgmt: MyGAPI is not initialized")
}
close(obj.closeChan)
obj.wg.Wait()
obj.initialized = false // closed = true
return nil
}
// Run runs an embedded mgmt server.
func Run() error {
obj := &mgmt.Main{}
obj.Program = "libmgmt" // TODO: set on compilation
obj.Version = "0.0.1" // TODO: set on compilation
obj.TmpPrefix = true
obj.IdealClusterSize = -1
obj.ConvergedTimeout = -1
obj.Noop = false // FIXME: careful!
obj.GAPI = &MyGAPI{ // graph API
Name: "libmgmt", // TODO: set on compilation
Interval: 60 * 10, // arbitrarily change graph every 15 seconds
}
if err := obj.Init(); err != nil {
return err
}
// install the exit signal handler
exit := make(chan struct{})
defer close(exit)
go func() {
signals := make(chan os.Signal, 1)
signal.Notify(signals, os.Interrupt) // catch ^C
//signal.Notify(signals, os.Kill) // catch signals
signal.Notify(signals, syscall.SIGTERM)
select {
case sig := <-signals: // any signal will do
if sig == os.Interrupt {
log.Println("Interrupted by ^C")
obj.Exit(nil)
return
}
log.Println("Interrupted by signal")
obj.Exit(fmt.Errorf("Killed by %v", sig))
return
case <-exit:
return
}
}()
if err := obj.Run(); err != nil {
return err
}
return nil
}
func main() {
log.Printf("Hello!")
if err := Run(); err != nil {
fmt.Println(err)
os.Exit(1)
return
}
log.Printf("Goodbye!")
}

View File

@@ -386,6 +386,7 @@ func (obj *Main) Run() error {
G.AutoEdges() // add autoedges; modifies the graph
G.AutoGroup() // run autogroup; modifies the graph
// TODO: do we want to do a transitive reduction?
// FIXME: run a type checker that verifies all the send->recv relationships
log.Printf("Graph: %v", G) // show graph
if obj.GraphvizFilter != "" {

View File

@@ -659,6 +659,14 @@ func (g *Graph) Process(v *Vertex) error {
}
obj.SetState(resources.ResStateCheckApply)
// connect any senders to receivers and detect if values changed
if changed, err := obj.SendRecv(obj); err != nil {
return errwrap.Wrapf(err, "could not SendRecv in Process")
} else if changed {
obj.StateOK(false) // invalidate cache
}
// if this fails, don't UpdateTimestamp()
checkok, err := obj.CheckApply(!obj.Meta().Noop)
if checkok && err != nil { // should never return this way

View File

@@ -198,6 +198,10 @@ func (obj *NspawnRes) CheckApply(apply bool) (checkok bool, err error) {
log.Printf("%s[%s]: CheckApply(%t)", obj.Kind(), obj.GetName(), apply)
}
if obj.isStateOK { // cache the state
return true, nil
}
// this resource depends on systemd ensure that it's running
if !systemdUtil.IsRunningSystemd() {
return false, errors.New("Systemd is not running.")

309
resources/password.go Normal file
View File

@@ -0,0 +1,309 @@
// 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 (
"crypto/rand"
"encoding/gob"
"fmt"
"log"
"math/big"
"os"
"path"
"strings"
"time"
"github.com/purpleidea/mgmt/event"
errwrap "github.com/pkg/errors"
)
func init() {
gob.Register(&PasswordRes{})
}
const (
alphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
newline = "\n" // something not in alphabet that TrimSpace can trim
)
// PasswordRes is a no-op resource that returns a random password string.
type PasswordRes struct {
BaseRes `yaml:",inline"`
// FIXME: is uint16 too big?
Length uint16 `yaml:"length"` // number of characters to return
Password *string // the generated password
path string // the path to local storage
}
// NewPasswordRes is a constructor for this resource. It also calls Init() for you.
func NewPasswordRes(name string, length uint16) (*PasswordRes, error) {
obj := &PasswordRes{
BaseRes: BaseRes{
Name: name,
},
Length: length,
}
return obj, obj.Init()
}
func (obj *PasswordRes) read() (string, error) {
file, err := os.Open(obj.path) // open a handle to read the file
if err != nil {
return "", errwrap.Wrapf(err, "could not read password")
}
defer file.Close()
data := make([]byte, obj.Length+uint16(len(newline))) // data + newline
if _, err := file.Read(data); err != nil {
return "", errwrap.Wrapf(err, "could not read from file")
}
return strings.TrimSpace(string(data)), nil
}
func (obj *PasswordRes) write(password string) (int, error) {
file, err := os.Create(obj.path) // open a handle to create the file
if err != nil {
return -1, errwrap.Wrapf(err, "can't create file")
}
defer file.Close()
return file.Write([]byte(password + newline))
}
// generate generates a new password.
func (obj *PasswordRes) generate() (string, error) {
max := len(alphabet) - 1 // last index
output := ""
// FIXME: have someone verify this is cryptographically secure & correct
for i := uint16(0); i < obj.Length; i++ {
big, err := rand.Int(rand.Reader, big.NewInt(int64(max)))
if err != nil {
return "", errwrap.Wrapf(err, "could not generate password")
}
ix := big.Int64()
output += string(alphabet[ix])
}
if output == "" { // safety against empty passwords
return "", fmt.Errorf("password is empty")
}
if uint16(len(output)) != obj.Length { // safety against weird bugs
return "", fmt.Errorf("password length is too short") // bug!
}
return output, nil
}
// check validates a stored password string
func (obj *PasswordRes) check(value string) error {
length := uint16(len(value))
if length != obj.Length {
return fmt.Errorf("String length is not %d", obj.Length)
}
Loop:
for i := uint16(0); i < length; i++ {
for j := 0; j < len(alphabet); j++ {
if value[i] == alphabet[j] {
continue Loop
}
}
// we couldn't find that character, so error!
return fmt.Errorf("Invalid character `%s`", string(value[i]))
}
return nil
}
// Init generates a new password for this resource if one was not provided. It
// will save this into a local file. It will load it back in from previous runs.
func (obj *PasswordRes) Init() error {
// XXX: eventually store a hash instead of the plain text! we might want
// to generate a new value on fresh run if the downstream resource needs
// an update (triggers a backpoke?) this is a POC for send/recv for now.
obj.BaseRes.kind = "Password" // must be set before using VarDir
dir, err := obj.VarDir("")
if err != nil {
return errwrap.Wrapf(err, "could not get VarDir in Init()")
}
obj.path = path.Join(dir, "password") // return a unique file
password := ""
if _, err := os.Stat(obj.path); err != nil { // probably doesn't exist
if !os.IsNotExist(err) {
return errwrap.Wrapf(err, "unknown stat error")
}
// generate password and store it in the file
if obj.Password != nil {
password = *obj.Password // reuse what we've got
} else {
var err error
if password, err = obj.generate(); err != nil { // generate one!
return errwrap.Wrapf(err, "could not init password")
}
}
// store it to disk
if _, err := obj.write(password); err != nil {
return errwrap.Wrapf(err, "can't write to file")
}
} else { // must exist already!
password, err := obj.read()
if err != nil {
return errwrap.Wrapf(err, "could not read password")
}
if err := obj.check(password); err != nil {
return errwrap.Wrapf(err, "check failed")
}
if p := obj.Password; p != nil && *p != password {
// stored password isn't consistent with memory
if _, err := obj.write(*p); err != nil {
return errwrap.Wrapf(err, "consistency overwrite failed")
}
password = *p // use the copy from the resource
}
}
obj.Password = &password // save in memory
return obj.BaseRes.Init() // call base init, b/c we're overriding
}
// Validate if the params passed in are valid data.
// FIXME: where should this get called ?
func (obj *PasswordRes) Validate() error {
return nil
}
// Watch is the primary listener for this resource and it outputs events.
func (obj *PasswordRes) Watch(processChan chan event.Event) error {
if obj.IsWatching() {
return nil // TODO: should this be an error?
}
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
}
var send = false // send event?
var exit = false
for {
obj.SetState(ResStateWatching) // reset
select {
case event := <-obj.Events():
cuid.SetConverged(false)
// we avoid sending events on unpause
if exit, send = obj.ReadEvent(&event); exit {
return nil // exit
}
case <-cuid.ConvergedTimer():
cuid.SetConverged(true) // converged!
continue
case <-Startup(startup):
cuid.SetConverged(false)
send = true
}
// do all our event sending all together to avoid duplicate msgs
if send {
startup = true // startup finished
send = false
// only do this on certain types of events
//obj.isStateOK = false // something made state dirty
if exit, err := obj.DoSend(processChan, ""); exit || err != nil {
return err // we exit or bubble up a NACK...
}
}
}
}
// CheckApply method for Password resource. Does nothing, returns happy!
func (obj *PasswordRes) CheckApply(apply bool) (checkok bool, err error) {
log.Printf("%s[%s]: CheckApply(%t)", obj.Kind(), obj.GetName(), apply)
return true, nil
}
// PasswordUID is the UID struct for PasswordRes.
type PasswordUID struct {
BaseUID
name string
}
// AutoEdges returns the AutoEdge interface. In this case no autoedges are used.
func (obj *PasswordRes) AutoEdges() AutoEdge {
return 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 *PasswordRes) GetUIDs() []ResUID {
x := &PasswordUID{
BaseUID: BaseUID{name: obj.GetName(), kind: obj.Kind()},
name: obj.Name,
}
return []ResUID{x}
}
// GroupCmp returns whether two resources can be grouped together or not.
func (obj *PasswordRes) GroupCmp(r Res) bool {
_, ok := r.(*PasswordRes)
if !ok {
// NOTE: technically we could group a noop into any other
// resource, if that resource knew how to handle it, although,
// since the mechanics of inter-kind resource grouping are
// tricky, avoid doing this until there's a good reason.
return false
}
return true // noop resources can always be grouped together!
}
// Compare two resources and return if they are equivalent.
func (obj *PasswordRes) Compare(res Res) bool {
switch res.(type) {
// we can only compare PasswordRes to others of the same resource
case *PasswordRes:
res := res.(*PasswordRes)
if !obj.BaseRes.Compare(res) { // call base Compare
return false
}
if obj.Name != res.Name {
return false
}
default:
return false
}
return true
}

View File

@@ -133,6 +133,9 @@ type Base interface {
DoSend(chan event.Event, string) (bool, error)
SendEvent(event.EventName, bool, bool) bool
ReadEvent(*event.Event) (bool, bool) // TODO: optional here?
SendRecv(Res) (bool, error) // send->recv data passing function
IsStateOK() bool
StateOK(b bool)
GroupCmp(Res) bool // TODO: is there a better name for this?
GroupRes(Res) error // group resource (arg) into self
IsGrouped() bool // am I grouped?
@@ -159,9 +162,12 @@ type Res interface {
type BaseRes struct {
Name string `yaml:"name"`
MetaParams MetaParams `yaml:"meta"` // struct of all the metaparams
Recv map[string]Send // mapping of key to receive on from value
kind string
events chan event.Event
converger converger.Converger // converged tracking
prefix string // base prefix for this resource
state ResState
watching bool // is Watch() loop running ?
isStateOK bool // whether the state is okay based on events or not
@@ -359,6 +365,16 @@ func (obj *BaseRes) ReadEvent(ev *event.Event) (exit, poke bool) {
return true, false // required to keep the stupid go compiler happy
}
// IsStateOK returns the cached state value.
func (obj *BaseRes) IsStateOK() bool {
return obj.isStateOK
}
// StateOK sets the cached state value.
func (obj *BaseRes) StateOK(b bool) {
obj.isStateOK = b
}
// GroupCmp compares two resources and decides if they're suitable for grouping
// You'll probably want to override this method when implementing a resource...
func (obj *BaseRes) GroupCmp(res Res) bool {

115
resources/sendrecv.go Normal file
View File

@@ -0,0 +1,115 @@
// 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 (
"fmt"
"log"
"reflect"
"github.com/purpleidea/mgmt/global"
multierr "github.com/hashicorp/go-multierror"
errwrap "github.com/pkg/errors"
)
// Send points to a value that a resource will send.
type Send struct {
Res Res // a handle to the resource which is sending a value
Key string // the key in the resource that we're sending
}
// SendRecv pulls in the sent values into the receive slots. It is called by the
// receiver and must be given as input the full resource struct to receive on.
func (obj *BaseRes) SendRecv(res Res) (bool, error) {
log.Printf("%s[%s]: SendRecv...", obj.Kind(), obj.GetName())
if global.DEBUG {
log.Printf("%s[%s]: SendRecv: Debug: %+v", obj.Kind(), obj.GetName(), obj.Recv)
}
var changed bool // did we update a value?
var err error
for k, v := range obj.Recv {
log.Printf("SendRecv: %s[%s].%s <- %s[%s].%s", obj.Kind(), obj.GetName(), k, v.Res.Kind(), v.Res.GetName(), v.Key)
// send
obj1 := reflect.Indirect(reflect.ValueOf(v.Res))
type1 := obj1.Type()
value1 := obj1.FieldByName(v.Key)
kind1 := value1.Kind()
// recv
obj2 := reflect.Indirect(reflect.ValueOf(res)) // pass in full struct
type2 := obj2.Type()
value2 := obj2.FieldByName(k)
kind2 := value2.Kind()
if global.DEBUG {
log.Printf("Send(%s) has %v: %v", type1, kind1, value1)
log.Printf("Recv(%s) has %v: %v", type2, kind2, value2)
}
// i think we probably want the same kind, at least for now...
if kind1 != kind2 {
e := fmt.Errorf("Kind mismatch between %s[%s]: %s and %s[%s]: %s", obj.Kind(), obj.GetName(), kind2, v.Res.Kind(), v.Res.GetName(), kind1)
err = multierr.Append(err, e) // list of errors
continue
}
// if the types don't match, we can't use send->recv
// TODO: do we want to relax this for string -> *string ?
if e := TypeCmp(value1, value2); e != nil {
e := errwrap.Wrapf(e, "Type mismatch between %s[%s] and %s[%s]", obj.Kind(), obj.GetName(), v.Res.Kind(), v.Res.GetName())
err = multierr.Append(err, e) // list of errors
continue
}
// if we can't set, then well this is pointless!
if !value2.CanSet() {
e := fmt.Errorf("Can't set %s[%s].%s", obj.Kind(), obj.GetName(), k)
err = multierr.Append(err, e) // list of errors
continue
}
// if we can't interface, we can't compare...
if !value1.CanInterface() || !value2.CanInterface() {
e := fmt.Errorf("Can't interface %s[%s].%s", obj.Kind(), obj.GetName(), k)
err = multierr.Append(err, e) // list of errors
continue
}
// if the values aren't equal, we're changing the receiver
if !reflect.DeepEqual(value1.Interface(), value2.Interface()) {
// TODO: can we catch the panics here in case they happen?
value2.Set(value1) // do it for all types that match
changed = true
}
}
return changed, err
}
// TypeCmp compares two reflect values to see if they are the same Kind. It can
// look into a ptr Kind to see if the underlying pair of ptr's can TypeCmp too!
func TypeCmp(a, b reflect.Value) error {
ta, tb := a.Type(), b.Type()
if ta != tb {
return fmt.Errorf("Type mismatch: %s != %s", ta, tb)
}
// NOTE: it seems we don't need to recurse into pointers to sub check!
return nil // identical Type()'s
}

View File

@@ -301,6 +301,7 @@ func (obj *SvcRes) CheckApply(apply bool) (checkok bool, err error) {
var startupOK = true // XXX: DETECT AND SET
if stateOK && startupOK {
obj.isStateOK = true
return true, nil // we are in the correct state
}
@@ -348,6 +349,7 @@ func (obj *SvcRes) CheckApply(apply bool) (checkok bool, err error) {
// XXX: also set enabled on boot
obj.isStateOK = true
return false, nil // success
}

View File

@@ -114,7 +114,7 @@ func (obj *TimerRes) Watch(processChan chan event.Event) error {
if send {
startup = true // startup finished
send = false
obj.isStateOK = false
//obj.isStateOK = false
if exit, err := obj.DoSend(processChan, "timer ticked"); exit || err != nil {
return err // we exit or bubble up a NACK...
}

View File

@@ -62,6 +62,7 @@ type Resources struct {
Msg []*resources.MsgRes `yaml:"msg"`
Noop []*resources.NoopRes `yaml:"noop"`
Nspawn []*resources.NspawnRes `yaml:"nspawn"`
Password []*resources.PasswordRes `yaml:"password"`
Pkg []*resources.PkgRes `yaml:"pkg"`
Svc []*resources.SvcRes `yaml:"svc"`
Timer []*resources.TimerRes `yaml:"timer"`