// Mgmt // Copyright (C) 2013-2018+ James Shubin and the project contributors // Written by James Shubin and the project contributors // // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU 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 General Public License for more details. // // You should have received a copy of the GNU General Public License // along with this program. If not, see . package resources import ( "fmt" "log" "reflect" "github.com/purpleidea/mgmt/event" multierr "github.com/hashicorp/go-multierror" errwrap "github.com/pkg/errors" ) // Event sends off an event, but doesn't block the incoming event queue. func (obj *BaseRes) Event() error { resp := event.NewResp() obj.processLock.Lock() if obj.processDone { obj.processLock.Unlock() return fmt.Errorf("processChan is already closed") } obj.quiesceGroup.Add(1) // add to processChan queue count obj.processChan <- &event.Event{Kind: event.EventNil, Resp: resp} // trigger process obj.processLock.Unlock() return resp.Wait() } // SendEvent pushes an event into the message queue for a particular vertex. func (obj *BaseRes) SendEvent(ev event.Kind, err error) error { if obj.debug { if err == nil { log.Printf("%s: SendEvent(%+v)", obj, ev) } else { log.Printf("%s: SendEvent(%+v): %v", obj, ev, err) } } resp := event.NewResp() obj.eventsLock.Lock() if obj.eventsDone { obj.eventsLock.Unlock() return fmt.Errorf("eventsChan is already closed") } obj.eventsChan <- &event.Event{Kind: ev, Resp: resp, Err: err} if ev == event.EventExit { obj.eventsDone = true close(obj.eventsChan) // this is where we properly close this channel! } obj.eventsLock.Unlock() resp.ACKWait() // waits until true (nil) value return nil } // ReadEvent processes events when a select gets one, and handles the pause // code too! The return values specify if we should exit and poke respectively. func (obj *BaseRes) ReadEvent(ev *event.Event) (exit *error, send bool) { //ev.ACK() err := ev.Error() switch ev.Kind { case event.EventStart: ev.ACK() return nil, true case event.EventPoke: ev.ACK() return nil, true case event.EventBackPoke: ev.ACK() return nil, true // forward poking in response to a back poke! case event.EventExit: obj.quiescing = true obj.quiesceGroup.Wait() obj.quiescing = false // for symmetry ev.ACK() // FIXME: what do we do if we have a pending refresh (poke) and an exit? return &err, false case event.EventPause: obj.quiescing = true // set the quiesce flag to avoid event replays obj.quiesceGroup.Wait() obj.quiescing = false // reset ev.ACK() // wait for next event to continue, but discard any backpoking! for { // Consider a graph (V2->V3). If while paused, we add a // new resource (V1->V2), when we unpause, V3 will run, // and then V2 followed by V1 (reverse topo sort) which // can cause V2 to BackPoke to V1 (since V1 needs to go // first) which can panic if V1 is not running yet! The // solution is to ignore the BackPoke because once that // V1 vertex gets running, it will then send off a poke // to V2 that it did without the need for the BackPoke! select { case e, ok := <-obj.Events(): if !ok { // shutdown err := error(nil) return &err, false } //obj.quiescing = true //obj.quiesceGroup.Wait() // unnecessary, but symmetrically correct //obj.quiescing = false e.ACK() err := e.Error() if e.Kind == event.EventExit { return &err, false } else if e.Kind == event.EventStart { // eventContinue return nil, false // don't poke on unpause! } else if e.Kind == event.EventBackPoke { continue // silently discard this event while paused } // if we get a poke event here, it's a bug! err = fmt.Errorf("%s: unknown event: %v, while paused", obj, e) panic(err) // TODO: return a special sentinel instead? //return &err, false } } } err = fmt.Errorf("unknown event: %v", ev) panic(err) // TODO: return a special sentinel instead? //return &err, false } // Running is called by the Watch method of the resource once it has started up. // This signals to the engine to kick off the initial CheckApply resource check. func (obj *BaseRes) Running() error { // TODO: If a non-polling resource wants to use the converger, then it // should probably tell Running (via an arg) to not do this. Currently // it's a very unlikely race that could cause an early converge if the // converge timeout is very short ( ~ 1s) and the Watch method doesn't // immediately SetConverged(false) to stop possible early termination. if obj.Meta().Poll == 0 { // if not polling, unblock this... obj.cuid.SetConverged(true) // a reasonable initial assumption } obj.StateOK(false) // assume we're initially dirty if !obj.isStarted { // this avoids a double close when/if watch retries obj.isStarted = true close(obj.started) // send started signal } var err error if obj.starter { // vertices of indegree == 0 should send initial pokes err = obj.Event() // trigger a CheckApply } return err // bubble up any possible error (or nil) } // 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 Changed bool // set to true if this key was updated, read only! } // SetRecv sets the Res Recv field to given map of Send structs. func (obj *BaseRes) SetRecv(recv map[string]*Send) { obj.Recv = recv } // 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) (map[string]bool, error) { if obj.debug { // NOTE: this could expose private resource data like passwords log.Printf("%s: SendRecv: %+v", obj, obj.Recv) } var updated = make(map[string]bool) // list of updated keys var err error for k, v := range obj.Recv { updated[k] = false // default v.Changed = false // reset to the default // 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 obj.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 and %s: %s", v.Res, kind1, obj, kind2) err = multierr.Append(err, e) // list of errors continue } // if the types don't match, we can't use send->recv // FIXME: do we want to relax this for string -> *string ? if e := TypeCmp(value1, value2); e != nil { e := errwrap.Wrapf(e, "type mismatch between %s and %s", v.Res, obj) 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", obj, 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", obj, 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 updated[k] = true // we updated this key! v.Changed = true // tag this key as updated! log.Printf("SendRecv: %s.%s -> %s.%s", v.Res, v.Key, obj, k) } } return updated, 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 }