gapi: langpuppet: Add initial implementation

This new entrypoint allows graph generation from both a Puppet manifest
and a piece of mcl code. The GAPI implementation wraps the two existing
GAPIs.

examples/langpuppet/graph1.mcl

@@ -0,0 +1,9 @@
+noop "puppet_first_handover" {}
+noop "puppet_second_handover" {}
+
+print "first message" {}
+print "third message" {}
+
+Print["first message"] -> Noop["puppet_first_handover"]
+
+Noop["puppet_second_handover"] -> Print["third message"]

examples/langpuppet/graph1.pp

@@ -0,0 +1,10 @@
+class mgmt_first_handover {}
+class mgmt_second_handover {}
+
+include mgmt_first_handover, mgmt_second_handover
+
+Class["mgmt_first_handover"]
+->
+notify { "second message": }
+->
+Class["mgmt_second_handover"]

langpuppet/gapi.go

@@ -0,0 +1,324 @@
+// Mgmt
+// Copyright (C) 2013-2018+ 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 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 <http://www.gnu.org/licenses/>.
+
+package langpuppet
+
+import (
+	"flag"
+	"fmt"
+	"strings"
+	"sync"
+
+	"github.com/purpleidea/mgmt/engine"
+	"github.com/purpleidea/mgmt/gapi"
+	"github.com/purpleidea/mgmt/lang"
+	"github.com/purpleidea/mgmt/pgraph"
+	"github.com/purpleidea/mgmt/puppet"
+
+	multierr "github.com/hashicorp/go-multierror"
+	errwrap "github.com/pkg/errors"
+	"github.com/urfave/cli"
+)
+
+const (
+	// Name is the name of this frontend.
+	Name = "langpuppet"
+	// FlagPrefix gets prepended to each flag of both the puppet and lang GAPI.
+	FlagPrefix = "lp-"
+)
+
+func init() {
+	gapi.Register(Name, func() gapi.GAPI { return &GAPI{} }) // register
+}
+
+// GAPI implements the main langpuppet GAPI interface.
+// It wraps the Puppet and Lang GAPIs and receives graphs from both.
+// It then runs a merging algorithm that mainly just makes a union
+// of both the sets of vertices and edges. Some vertices are merged
+// using a naming convention. Details can be found in the
+// langpuppet.mergeGraphs function.
+type GAPI struct {
+	langGAPI   gapi.GAPI // the wrapped lang entrypoint
+	puppetGAPI gapi.GAPI // the wrapped puppet entrypoint
+
+	currentLangGraph   *pgraph.Graph // the most recent graph received from lang
+	currentPuppetGraph *pgraph.Graph // the most recent graph received from puppet
+
+	langGraphReady   bool // flag to indicate that a new graph from lang is ready
+	puppetGraphReady bool // flag to indicate that a new graph from puppet is ready
+	graphFlagMutex   *sync.Mutex
+
+	data        gapi.Data
+	initialized bool
+	closeChan   chan struct{}
+	wg          sync.WaitGroup // sync group for tunnel go routines
+}
+
+// Cli takes a cli.Context, and returns our GAPI if activated. All arguments
+// should take the prefix of the registered name. On activation, if there are
+// any validation problems, you should return an error. If this was not
+// activated, then you should return a nil GAPI and a nil error.
+func (obj *GAPI) Cli(c *cli.Context, fs engine.Fs) (*gapi.Deploy, error) {
+	if !c.IsSet(FlagPrefix+lang.Name) && !c.IsSet(FlagPrefix+puppet.Name) {
+		return nil, nil
+	}
+
+	if !c.IsSet(FlagPrefix+lang.Name) || c.String(FlagPrefix+lang.Name) == "" {
+		return nil, fmt.Errorf("%s input is empty", FlagPrefix+lang.Name)
+	}
+	if !c.IsSet(FlagPrefix+puppet.Name) || c.String(FlagPrefix+puppet.Name) == "" {
+		return nil, fmt.Errorf("%s input is empty", FlagPrefix+puppet.Name)
+	}
+
+	flagSet := flag.NewFlagSet(Name, flag.ContinueOnError)
+
+	for _, flag := range c.FlagNames() {
+		if !c.IsSet(flag) {
+			continue
+		}
+		childFlagName := strings.TrimPrefix(flag, FlagPrefix)
+		flagSet.String(childFlagName, "", "no usage string needed here")
+		flagSet.Set(childFlagName, c.String(flag))
+	}
+
+	var langDeploy *gapi.Deploy
+	var puppetDeploy *gapi.Deploy
+	var err error
+
+	// we don't really need the deploy object from the child GAPIs
+	if langDeploy, err = (&lang.GAPI{}).Cli(cli.NewContext(c.App, flagSet, nil), fs); err != nil {
+		return nil, err
+	}
+	if puppetDeploy, err = (&puppet.GAPI{}).Cli(cli.NewContext(c.App, flagSet, nil), fs); err != nil {
+		return nil, err
+	}
+
+	return &gapi.Deploy{
+		Name: Name,
+		Noop: c.GlobalBool("noop"),
+		Sema: c.GlobalInt("sema"),
+		GAPI: &GAPI{
+			langGAPI:   langDeploy.GAPI,
+			puppetGAPI: puppetDeploy.GAPI,
+		},
+	}, nil
+}
+
+// CliFlags returns a list of flags used by this deploy subcommand.
+// It consists of all flags accepted by lang and puppet mode,
+// with a respective "lp-" prefix.
+func (obj *GAPI) CliFlags() []cli.Flag {
+	langFlags := (&lang.GAPI{}).CliFlags()
+	puppetFlags := (&puppet.GAPI{}).CliFlags()
+
+	var childFlags []cli.Flag
+	for _, flag := range append(langFlags, puppetFlags...) {
+		childFlags = append(childFlags, &cli.StringFlag{
+			Name:  FlagPrefix + strings.Split(flag.GetName(), ",")[0],
+			Value: "",
+			Usage: fmt.Sprintf("equivalent for '%s' when using the lang/puppet entrypoint", flag.GetName()),
+		})
+	}
+
+	return childFlags
+}
+
+// Init initializes the langpuppet GAPI struct.
+func (obj *GAPI) Init(data gapi.Data) error {
+	if obj.initialized {
+		return fmt.Errorf("already initialized")
+	}
+	obj.data = data // store for later
+	obj.graphFlagMutex = &sync.Mutex{}
+
+	dataLang := gapi.Data{
+		Program:       obj.data.Program,
+		Hostname:      obj.data.Hostname,
+		World:         obj.data.World,
+		Noop:          obj.data.Noop,
+		NoConfigWatch: obj.data.NoConfigWatch,
+		NoStreamWatch: obj.data.NoStreamWatch,
+		Debug:         obj.data.Debug,
+		Logf: func(format string, v ...interface{}) {
+			obj.data.Logf(lang.Name+": "+format, v...)
+		},
+	}
+	dataPuppet := gapi.Data{
+		Program:       obj.data.Program,
+		Hostname:      obj.data.Hostname,
+		World:         obj.data.World,
+		Noop:          obj.data.Noop,
+		NoConfigWatch: obj.data.NoConfigWatch,
+		NoStreamWatch: obj.data.NoStreamWatch,
+		Debug:         obj.data.Debug,
+		Logf: func(format string, v ...interface{}) {
+			obj.data.Logf(puppet.Name+": "+format, v...)
+		},
+	}
+
+	if err := obj.langGAPI.Init(dataLang); err != nil {
+		return err
+	}
+	if err := obj.puppetGAPI.Init(dataPuppet); err != nil {
+		return err
+	}
+
+	obj.closeChan = make(chan struct{})
+	obj.initialized = true
+	return nil
+}
+
+// Graph returns a current Graph.
+func (obj *GAPI) Graph() (*pgraph.Graph, error) {
+	if !obj.initialized {
+		return nil, fmt.Errorf("%s: GAPI is not initialized", Name)
+	}
+
+	var err error
+	obj.graphFlagMutex.Lock()
+	if obj.langGraphReady {
+		obj.langGraphReady = false
+		obj.graphFlagMutex.Unlock()
+		obj.currentLangGraph, err = obj.langGAPI.Graph()
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		obj.graphFlagMutex.Unlock()
+	}
+
+	obj.graphFlagMutex.Lock()
+	if obj.puppetGraphReady {
+		obj.puppetGraphReady = false
+		obj.graphFlagMutex.Unlock()
+		obj.currentPuppetGraph, err = obj.puppetGAPI.Graph()
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		obj.graphFlagMutex.Unlock()
+	}
+
+	g, err := mergeGraphs(obj.currentLangGraph, obj.currentPuppetGraph)
+
+	if obj.data.Debug {
+		obj.currentLangGraph.Logf(func(format string, v ...interface{}) {
+			obj.data.Logf("graph: "+lang.Name+": "+format, v...)
+		})
+		obj.currentPuppetGraph.Logf(func(format string, v ...interface{}) {
+			obj.data.Logf("graph: "+puppet.Name+": "+format, v...)
+		})
+		if err == nil {
+			g.Logf(func(format string, v ...interface{}) {
+				obj.data.Logf("graph: "+Name+": "+format, v...)
+			})
+		}
+	}
+
+	return g, err
+}
+
+// Next returns nil errors every time there could be a new graph.
+func (obj *GAPI) Next() chan gapi.Next {
+	ch := make(chan gapi.Next)
+	obj.wg.Add(1)
+	go func() {
+		defer obj.wg.Done()
+		defer close(ch) // this will run before the obj.wg.Done()
+		if !obj.initialized {
+			next := gapi.Next{
+				Err:  fmt.Errorf("%s: GAPI is not initialized", Name),
+				Exit: true, // exit, b/c programming error?
+			}
+			ch <- next
+			return
+		}
+		nextLang := obj.langGAPI.Next()
+		nextPuppet := obj.puppetGAPI.Next()
+
+		firstLang := false
+		firstPuppet := false
+
+		for {
+			var err error
+			exit := false
+			select {
+			case nextChild := <-nextLang:
+				if nextChild.Err != nil {
+					err = nextChild.Err
+					exit = nextChild.Exit
+				} else {
+					obj.graphFlagMutex.Lock()
+					obj.langGraphReady = true
+					obj.graphFlagMutex.Unlock()
+					firstLang = true
+				}
+			case nextChild := <-nextPuppet:
+				if nextChild.Err != nil {
+					err = nextChild.Err
+					exit = nextChild.Exit
+				} else {
+					obj.graphFlagMutex.Lock()
+					obj.puppetGraphReady = true
+					obj.graphFlagMutex.Unlock()
+					firstPuppet = true
+				}
+			case <-obj.closeChan:
+				return
+			}
+
+			if (!firstLang || !firstPuppet) && err == nil {
+				continue
+			}
+
+			if err == nil {
+				obj.data.Logf("generating new composite graph...")
+			}
+			next := gapi.Next{
+				Exit: exit,
+				Err:  err,
+			}
+
+			select {
+			case ch <- next: // trigger a run (send a msg)
+			// unblock if we exit while waiting to send!
+			case <-obj.closeChan:
+				return
+			}
+		}
+	}()
+	return ch
+}
+
+// Close shuts down the Puppet GAPI.
+func (obj *GAPI) Close() error {
+	if !obj.initialized {
+		return fmt.Errorf("%s: GAPI is not initialized", Name)
+	}
+
+	var err error
+	if e := obj.langGAPI.Close(); e != nil {
+		err = multierr.Append(err, errwrap.Wrapf(e, "closing lang GAPI failed"))
+	}
+	if e := obj.puppetGAPI.Close(); e != nil {
+		err = multierr.Append(err, errwrap.Wrapf(e, "closing Puppet GAPI failed"))
+	}
+
+	close(obj.closeChan)
+	obj.initialized = false // closed = true
+	return err
+}

langpuppet/merge.go

@@ -0,0 +1,169 @@
+// Mgmt
+// Copyright (C) 2013-2018+ 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 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 <http://www.gnu.org/licenses/>.
+
+// Package langpuppet implements an integration entrypoint that combines lang and Puppet.
+package langpuppet
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/purpleidea/mgmt/engine"
+	"github.com/purpleidea/mgmt/pgraph"
+)
+
+const (
+	// MergePrefixLang is how a mergeable vertex name starts in mcl code.
+	MergePrefixLang = "puppet_"
+	// MergePrefixPuppet is how a mergeable Puppet class name starts.
+	MergePrefixPuppet = "mgmt_"
+)
+
+// mergeGraph returns the merged graph containing all vertices
+// and edges found in the graphs produced by the lang and Puppet GAPIs
+// associated with the wrapping GAPI.
+// Vertices are merged if they adhere to the following rules (for any
+// given value of POSTFIX):
+//  1. The graph from lang contains a noop vertex named puppet_POSTFIX
+//  2. The graph from Puppet contains an empty class mgmt_POSTFIX
+//  3. The resulting graph will contain one noop vertex named POSTFIX
+//     that replaces all nodes mentioned in (1) and (2)
+// All edges connecting to any of the vertices merged this way
+// will be present in the merged graph.
+func mergeGraphs(graphFromLang, graphFromPuppet *pgraph.Graph) (*pgraph.Graph, error) {
+	if graphFromLang == nil || graphFromPuppet == nil {
+		return nil, fmt.Errorf("cannot merge graphs until both child graphs are loaded")
+	}
+
+	result, err := pgraph.NewGraph(graphFromLang.Name + "+" + graphFromPuppet.Name)
+	if err != nil {
+		return nil, err
+	}
+
+	mergeTargets := make(map[string]pgraph.Vertex)
+
+	// first add all vertices from the lang graph
+	for _, vertex := range graphFromLang.Vertices() {
+		if strings.Index(vertex.String(), "noop["+MergePrefixLang) == 0 {
+			resource, ok := vertex.(engine.Res)
+			if !ok {
+				return nil, fmt.Errorf("vertex %s is not a named resource", vertex.String())
+			}
+			basename := strings.TrimPrefix(resource.Name(), MergePrefixLang)
+			resource.SetName(basename)
+			mergeTargets[basename] = vertex
+		}
+		result.AddVertex(vertex)
+		for _, neighbor := range graphFromLang.OutgoingGraphVertices(vertex) {
+			result.AddVertex(neighbor)
+			result.AddEdge(vertex, neighbor, graphFromLang.FindEdge(vertex, neighbor))
+		}
+	}
+
+	var anchor pgraph.Vertex
+	mergePairs := make(map[pgraph.Vertex]pgraph.Vertex)
+
+	// do a scan through the Puppet graph, and mark all vertices that will be
+	// subject to a merge, so it will be easier do generate the new edges
+	// in the final pass
+	for _, vertex := range graphFromPuppet.Vertices() {
+		if vertex.String() == "noop[admissible_Stage[main]]" {
+			// we can start a depth first search here
+			anchor = vertex
+			continue
+		}
+		// at this stage we don't distinguis between class start and end
+		if strings.Index(vertex.String(), "noop[admissible_Class["+strings.Title(MergePrefixPuppet)) != 0 &&
+			strings.Index(vertex.String(), "noop[completed_Class["+strings.Title(MergePrefixPuppet)) != 0 {
+			continue
+		}
+
+		resource, ok := vertex.(engine.Res)
+		if !ok {
+			return nil, fmt.Errorf("vertex %s is not a named resource", vertex.String())
+		}
+		// strip either prefix (plus the closing bracket)
+		basename := strings.TrimSuffix(
+			strings.TrimPrefix(
+				strings.TrimPrefix(resource.Name(),
+					"admissible_Class["+strings.Title(MergePrefixPuppet)),
+				"completed_Class["+strings.Title(MergePrefixPuppet)),
+			"]")
+
+		if _, found := mergeTargets[basename]; !found {
+			// FIXME: should be a warning not an error?
+			return nil, fmt.Errorf("puppet graph has unmatched class %s%s", MergePrefixPuppet, basename)
+		}
+
+		mergePairs[vertex] = mergeTargets[basename]
+
+		if strings.Index(resource.Name(), "admissible_Class["+strings.Title(MergePrefixPuppet)) != 0 {
+			continue
+		}
+
+		// is there more than one edge outgoing from the class start?
+		if graphFromPuppet.OutDegree()[vertex] > 1 {
+			return nil, fmt.Errorf("class %s is not empty", basename)
+		}
+
+		// does this edge not lead to the class end?
+		next := graphFromPuppet.OutgoingGraphVertices(vertex)[0]
+		if next.String() != "noop[completed_Class["+strings.Title(MergePrefixPuppet)+basename+"]]" {
+			return nil, fmt.Errorf("class %s%s is not empty, start is followed by %s", MergePrefixPuppet, basename, next.String())
+		}
+	}
+
+	merged := make(map[pgraph.Vertex]bool)
+	result.AddVertex(anchor)
+	// traverse the puppet graph, add all vertices and perform merges
+	// using DFS so we can be sure the "admissible" is visited before the "completed" vertex
+	for _, vertex := range graphFromPuppet.DFS(anchor) {
+		source := vertex
+
+		// when adding edges, the source might be a different vertex
+		// than the current one, if this is a merged vertex
+		if _, found := mergePairs[vertex]; found {
+			source = mergePairs[vertex]
+		}
+
+		// the current vertex has been added by previous iterations,
+		// we only add neighbors here
+		for _, neighbor := range graphFromPuppet.OutgoingGraphVertices(vertex) {
+			if strings.Index(neighbor.String(), "noop[admissible_Class["+strings.Title(MergePrefixPuppet)) == 0 {
+				result.AddEdge(source, mergePairs[neighbor], graphFromPuppet.FindEdge(vertex, neighbor))
+				continue
+			}
+			if strings.Index(neighbor.String(), "noop[completed_Class["+strings.Title(MergePrefixPuppet)) == 0 {
+				// mark target vertex as merged
+				merged[mergePairs[neighbor]] = true
+				continue
+			}
+			// if we reach here, this neighbor is a regular vertex
+			result.AddVertex(neighbor)
+			result.AddEdge(source, neighbor, graphFromPuppet.FindEdge(vertex, neighbor))
+		}
+	}
+
+	for _, vertex := range mergeTargets {
+		if !merged[vertex] {
+			// FIXME: should be a warning not an error?
+			return nil, fmt.Errorf("lang graph has unmatched %s", vertex.String())
+		}
+	}
+
+	return result, nil
+}

lib/deploy.go

@@ -27,6 +27,7 @@ import (
 	"github.com/purpleidea/mgmt/gapi"
 	// these imports are so that GAPIs register themselves in init()
 	_ "github.com/purpleidea/mgmt/lang"
+	_ "github.com/purpleidea/mgmt/langpuppet"
 	_ "github.com/purpleidea/mgmt/puppet"
 	_ "github.com/purpleidea/mgmt/yamlgraph"