git: Improve tagging script

test: Disable prometheus-3.sh test temporarily
It seems to be failing, and I'm not sure where the regression is, or if there is a race. Sorry roidelapluie.
2017-03-11 08:29:53 -05:00 · 2017-03-09 11:46:21 -05:00 · 2017-03-09 06:35:15 -05:00 · 2017-03-09 02:50:55 -05:00 · 2017-03-09 02:50:55 -05:00 · 2017-03-08 19:23:33 -05:00
56 changed files with 1914 additions and 675 deletions
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,9 +1,10 @@
 language: go
 go:
-  - 1.6
-  - 1.7
-  - 1.8
+  - 1.6.x
+  - 1.7.x
+  - 1.8.x
  - tip
+go_import_path: github.com/purpleidea/mgmt
 sudo: true
 dist: trusty
 before_install:
@@ -15,7 +16,7 @@ matrix:
  fast_finish: true
  allow_failures:
    - go: tip
-    - go: 1.8
+    - go: 1.8.x
 notifications:
  irc:
    channels:
--- a/4
+++ b/4
@@ -41,6 +41,8 @@ ifneq ($(GOTAGS),)
    BUILD_FLAGS = -tags '$(GOTAGS)'
 endif

+default: build
+
 #
 #	art
 #
@@ -110,7 +112,7 @@ ifneq ($(OLDGOLANG),)
 	@# avoid equals sign in old golang versions eg in: -X foo=bar
 	time go build -ldflags "-X main.program $(PROGRAM) -X main.version $(SVERSION)" -o $(PROGRAM) $(BUILD_FLAGS);
 else
-	time go build -ldflags "-X main.program=$(PROGRAM) -X main.version=$(SVERSION)" -o $(PROGRAM) $(BUILD_FLAGS);
+	time go build -i -ldflags "-X main.program=$(PROGRAM) -X main.version=$(SVERSION)" -o $(PROGRAM) $(BUILD_FLAGS);
 endif

 $(PROGRAM).static: main.go
--- a/README.md
+++ b/README.md
@@ -77,6 +77,7 @@ We'd love to have your patches! Please send them by email, or as a pull request.
 | James Shubin | video | [Recording from High Load Strategy 2016](https://vimeo.com/191493409) |
 | James Shubin | video | [Recording from NLUUG 2016](https://www.youtube.com/watch?v=MmpwOQAb_SE&html5=1) |
 | James Shubin | blog | [Send/Recv in mgmt](https://ttboj.wordpress.com/2016/12/07/sendrecv-in-mgmt/) |
+| James Shubin | blog | [Metaparameters in mgmt](https://ttboj.wordpress.com/2017/03/01/metaparameters-in-mgmt/) |

 ##

--- a/docs/documentation.md
+++ b/docs/documentation.md
@@ -24,6 +24,7 @@ For more information, you may like to read some blog posts from the author:
 * [Automatic clustering in mgmt](https://ttboj.wordpress.com/2016/06/20/automatic-clustering-in-mgmt/)
 * [Remote execution in mgmt](https://ttboj.wordpress.com/2016/10/07/remote-execution-in-mgmt/)
 * [Send/Recv in mgmt](https://ttboj.wordpress.com/2016/12/07/sendrecv-in-mgmt/)
+* [Metaparameters in mgmt](https://ttboj.wordpress.com/2017/03/01/metaparameters-in-mgmt/)

 There is also an [introductory video](http://meetings-archive.debian.net/pub/debian-meetings/2016/debconf16/Next_Generation_Config_Mgmt.webm) available.
 Older videos and other material [is available](https://github.com/purpleidea/mgmt/#on-the-web).
@@ -181,6 +182,7 @@ parameter with the [Noop](#Noop) resource.
 * [Exec](#Exec): Execute shell commands on the system.
 * [File](#File): Manage files and directories.
 * [Hostname](#Hostname): Manages the hostname on the system.
+* [KV](#KV): Set a key value pair in our shared world database.
 * [Msg](#Msg): Send log messages.
 * [Noop](#Noop): A simple resource that does nothing.
 * [Nspawn](#Nspawn): Manage systemd-machined nspawn containers.
@@ -267,6 +269,30 @@ The pretty hostname is a free-form UTF8 host name for presentation to the user.
 Hostname is the fallback value for all 3 fields above, if only `hostname` is
 specified, it will set all 3 fields to this value.

+### KV
+
+The KV resource sets a key and value pair in the global world database. This is
+quite useful for setting a flag after a number of resources have run. It will
+ignore database updates to the value that are greater in compare order than the
+requested key if the `SkipLessThan` parameter is set to true. If we receive a
+refresh, then the stored value will be reset to the requested value even if the
+stored value is greater.
+
+#### Key
+The string key used to store the key.
+
+#### Value
+The string value to set. This can also be set via Send/Recv.
+
+#### SkipLessThan
+If this parameter is set to `true`, then it will ignore updating the value as
+long as the database versions are greater than the requested value. The compare
+operation used is based on the `SkipCmpStyle` parameter.
+
+#### SkipCmpStyle
+By default this converts the string values to integers and compares them as you
+would expect.
+
 ### Msg

 The msg resource sends messages to the main log, or an external service such
@@ -457,6 +483,17 @@ the rate limiter as designated by the `Limit` value. If the `Limit` is not set
 to `+Infinity`, this must be a non-zero value. Please see the
 [rate](https://godoc.org/golang.org/x/time/rate) package for more information.

+#### Sema
+List of string ids. Sema is a P/V style counting semaphore which can be used to
+limit parallelism during the CheckApply phase of resource execution. Each
+resource can have `N` different semaphores which share a graph global namespace.
+Each semaphore has a maximum count associated with it. The default value of the
+size is 1 (one) if size is unspecified. Each string id is the unique id of the
+semaphore. If the id contains a trailing colon (:) followed by a positive
+integer, then that value is the max size for that semaphore. Valid semaphore
+id's include: `some_id`, `hello:42`, `not:smart:4` and `:13`. It is expected
+that the last bare example be only used by the engine to add a global semaphore.
+
 ### Graph definition file
 graph.yaml is the compiled graph definition file. The format is currently
 undocumented, but by looking through the [examples/](https://github.com/purpleidea/mgmt/tree/master/examples)
@@ -481,6 +518,15 @@ Globally force all resources into no-op mode. This also disables the export to
 etcd functionality, but does not disable resource collection, however all
 resources that are collected will have their individual noop settings set.

+#### `--sema <size>`
+Globally add a counting semaphore of this size to each resource in the graph.
+The semaphore will get given an id of `:size`. In other words if you specify a
+size of 42, you can expect a semaphore if named: `:42`. It is expected that
+consumers of the semaphore metaparameter always include a prefix to avoid a
+collision with this globally defined semaphore. The size value must be greater
+than zero at this time. The traditional non-parallel execution found in config
+management tools such as `Puppet` can be obtained with `--sema 1`.
+
 #### `--remote <graph.yaml>`
 Point to a graph file to run on the remote host specified within. This parameter
 can be used multiple times if you'd like to remotely run on multiple hosts in
--- a/docs/prometheus.md
+++ b/docs/prometheus.md
@@ -31,6 +31,7 @@ Here is a list of the metrics we provide:
 - `mgmt_checkapply_total`: The number of CheckApply's that mgmt has run
 - `mgmt_failures_total`: The number of resources that have failed
 - `mgmt_failures_current`: The number of resources that have failed
+- `mgmt_graph_start_time_seconds`: Start time of the current graph since unix epoch in seconds

 For each metric, you will get some extra labels:

--- a/docs/resource-guide.md
+++ b/docs/resource-guide.md
@@ -80,7 +80,7 @@ work, and finish by calling the `Init` method of the base resource.
 ```golang
 // Init initializes the Foo resource.
 func (obj *FooRes) Init() error {
-	obj.BaseRes.kind = "Foo" // must set capitalized resource kind
+	obj.BaseRes.kind = "foo" // must lower case resource kind
 	// run the resource specific initialization, and error if anything fails
 	if some_error {
 		return err // something went wrong!
--- a/etcd/str.go
+++ b/etcd/str.go
@@ -0,0 +1,115 @@
+// Mgmt
+// Copyright (C) 2013-2017+ 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 etcd
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/purpleidea/mgmt/util"
+
+	etcd "github.com/coreos/etcd/clientv3"
+	errwrap "github.com/pkg/errors"
+)
+
+// WatchStr returns a channel which spits out events on key activity.
+// FIXME: It should close the channel when it's done, and spit out errors when
+// something goes wrong.
+func WatchStr(obj *EmbdEtcd, key string) chan error {
+	// new key structure is /$NS/strings/$key/$hostname = $data
+	path := fmt.Sprintf("/%s/strings/%s", NS, key)
+	ch := make(chan error, 1)
+	// FIXME: fix our API so that we get a close event on shutdown.
+	callback := func(re *RE) error {
+		// TODO: is this even needed? it used to happen on conn errors
+		//log.Printf("Etcd: Watch: Path: %v", path) // event
+		if re == nil || re.response.Canceled {
+			return fmt.Errorf("watch is empty") // will cause a CtxError+retry
+		}
+		if len(ch) == 0 { // send event only if one isn't pending
+			ch <- nil // event
+		}
+		return nil
+	}
+	_, _ = obj.AddWatcher(path, callback, true, false, etcd.WithPrefix()) // no need to check errors
+	return ch
+}
+
+// GetStr collects all of the strings which match a namespace in etcd.
+func GetStr(obj *EmbdEtcd, hostnameFilter []string, key string) (map[string]string, error) {
+	// old key structure is /$NS/strings/$hostname/$key = $data
+	// new key structure is /$NS/strings/$key/$hostname = $data
+	// FIXME: if we have the $key as the last token (old key structure), we
+	// can allow the key to contain the slash char, otherwise we need to
+	// verify that one isn't present in the input string.
+	path := fmt.Sprintf("/%s/strings/%s", NS, key)
+	keyMap, err := obj.Get(path, etcd.WithPrefix(), etcd.WithSort(etcd.SortByKey, etcd.SortAscend))
+	if err != nil {
+		return nil, errwrap.Wrapf(err, "could not get strings in: %s", key)
+	}
+	result := make(map[string]string)
+	for key, val := range keyMap {
+		if !strings.HasPrefix(key, path) { // sanity check
+			continue
+		}
+
+		str := strings.Split(key[len(path):], "/")
+		if len(str) != 2 {
+			return nil, fmt.Errorf("unexpected chunk count of %d", len(str))
+		}
+		_, hostname := str[0], str[1]
+
+		if hostname == "" {
+			return nil, fmt.Errorf("unexpected chunk length of %d", len(hostname))
+		}
+
+		// FIXME: ideally this would be a server side filter instead!
+		if len(hostnameFilter) > 0 && !util.StrInList(hostname, hostnameFilter) {
+			continue
+		}
+		//log.Printf("Etcd: GetStr(%s): (Hostname, Data): (%s, %s)", key, hostname, val)
+		result[hostname] = val
+	}
+	return result, nil
+}
+
+// SetStr sets a key and hostname pair to a certain value. If the value is nil,
+// then it deletes the key. Otherwise the value should point to a string.
+// TODO: TTL or delete disconnect?
+func SetStr(obj *EmbdEtcd, hostname, key string, data *string) error {
+	// key structure is /$NS/strings/$key/$hostname = $data
+	path := fmt.Sprintf("/%s/strings/%s/%s", NS, key, hostname)
+	ifs := []etcd.Cmp{} // list matching the desired state
+	ops := []etcd.Op{}  // list of ops in this transaction (then)
+	els := []etcd.Op{}  // list of ops in this transaction (else)
+	if data == nil {    // perform a delete
+		// TODO: use https://github.com/coreos/etcd/pull/7417 if merged
+		//ifs = append(ifs, etcd.KeyExists(path))
+		ifs = append(ifs, etcd.Compare(etcd.Version(path), ">", 0))
+		ops = append(ops, etcd.OpDelete(path))
+	} else {
+		data := *data                                                // get the real value
+		ifs = append(ifs, etcd.Compare(etcd.Value(path), "=", data)) // desired state
+		els = append(els, etcd.OpPut(path, data))
+	}
+
+	// it's important to do this in one transaction, and atomically, because
+	// this way, we only generate one watch event, and only when it's needed
+	_, err := obj.Txn(ifs, ops, els) // TODO: do we need to look at response?
+	return errwrap.Wrapf(err, "could not set strings in: %s", key)
+}
--- a/etcd/world.go
+++ b/etcd/world.go
@@ -41,3 +41,24 @@ func (obj *World) ResCollect(hostnameFilter, kindFilter []string) ([]resources.R
 	// enforce that here if the underlying API supported it... Add this?
 	return GetResources(obj.EmbdEtcd, hostnameFilter, kindFilter)
 }
+
+// SetWatch returns a channel which spits out events on possible string changes.
+func (obj *World) StrWatch(namespace string) chan error {
+	return WatchStr(obj.EmbdEtcd, namespace)
+}
+
+// StrGet returns a map of hostnames to values in the given namespace.
+func (obj *World) StrGet(namespace string) (map[string]string, error) {
+	return GetStr(obj.EmbdEtcd, []string{}, namespace)
+}
+
+// StrSet sets the namespace value to a particular string under the identity of
+// its own hostname.
+func (obj *World) StrSet(namespace, value string) error {
+	return SetStr(obj.EmbdEtcd, obj.Hostname, namespace, &value)
+}
+
+// StrDel deletes the value in a particular namespace.
+func (obj *World) StrDel(namespace string) error {
+	return SetStr(obj.EmbdEtcd, obj.Hostname, namespace, nil)
+}
--- a/examples/exec3-sema.yaml
+++ b/examples/exec3-sema.yaml
@@ -0,0 +1,67 @@
+---
+graph: parallel
+resources:
+  exec:
+  - name: pkg10
+    meta:
+      sema: ['mylock:1', 'otherlock:42']
+    cmd: sleep 10s
+    shell: ''
+    timeout: 0
+    watchcmd: ''
+    watchshell: ''
+    ifcmd: ''
+    ifshell: ''
+    pollint: 0
+    state: present
+  - name: svc10
+    meta:
+      sema: ['mylock:1']
+    cmd: sleep 10s
+    shell: ''
+    timeout: 0
+    watchcmd: ''
+    watchshell: ''
+    ifcmd: ''
+    ifshell: ''
+    pollint: 0
+    state: present
+  - name: exec10
+    meta:
+      sema: ['mylock:1']
+    cmd: sleep 10s
+    shell: ''
+    timeout: 0
+    watchcmd: ''
+    watchshell: ''
+    ifcmd: ''
+    ifshell: ''
+    pollint: 0
+    state: present
+  - name: pkg15
+    meta:
+      sema: ['mylock:1', 'otherlock:42']
+    cmd: sleep 15s
+    shell: ''
+    timeout: 0
+    watchcmd: ''
+    watchshell: ''
+    ifcmd: ''
+    ifshell: ''
+    pollint: 0
+    state: present
+edges:
+- name: e1
+  from:
+    kind: exec
+    name: pkg10
+  to:
+    kind: exec
+    name: svc10
+- name: e2
+  from:
+    kind: exec
+    name: svc10
+  to:
+    kind: exec
+    name: exec10
--- a/examples/kv1.yaml
+++ b/examples/kv1.yaml
@@ -0,0 +1,8 @@
+---
+graph: mygraph
+resources:
+  kv:
+  - name: kv1
+    key: "hello"
+    value: "world"
+edges: []
--- a/examples/kv2.yaml
+++ b/examples/kv2.yaml
@@ -0,0 +1,7 @@
+---
+graph: mygraph
+resources:
+  kv:
+  - name: kv1
+    key: "iamdeleted"
+edges: []
--- a/examples/kv3.yaml
+++ b/examples/kv3.yaml
@@ -0,0 +1,9 @@
+---
+graph: mygraph
+resources:
+  kv:
+  - name: kv1
+    key: "stage"
+    value: "3"
+    skiplessthan: true
+edges: []
--- a/examples/kv4.yaml
+++ b/examples/kv4.yaml
@@ -0,0 +1,31 @@
+---
+graph: mygraph
+resources:
+  kv:
+  - name: kv1
+    key: "stage"
+    value: "1"
+    skiplessthan: true
+  - name: kv2
+    key: "stage"
+    value: "2"
+    skiplessthan: true
+  - name: kv3
+    key: "stage"
+    value: "3"
+    skiplessthan: true
+edges:
+- name: e1
+  from:
+    kind: kv
+    name: kv1
+  to:
+    kind: kv
+    name: kv2
+- name: e2
+  from:
+    kind: kv
+    name: kv2
+  to:
+    kind: kv
+    name: kv3
--- a/gapi/gapi.go
+++ b/gapi/gapi.go
@@ -23,19 +23,10 @@ import (
 	"github.com/purpleidea/mgmt/resources"
 )

-// World is an interface to the rest of the different graph state. It allows
-// the GAPI to store state and exchange information throughout the cluster. It
-// is the interface each machine uses to communicate with the rest of the world.
-type World interface { // TODO: is there a better name for this interface?
-	ResExport([]resources.Res) error
-	// FIXME: should this method take a "filter" data struct instead of many args?
-	ResCollect(hostnameFilter, kindFilter []string) ([]resources.Res, error)
-}
-
 // Data is the set of input values passed into the GAPI structs via Init.
 type Data struct {
 	Hostname string // uuid for the host, required for GAPI
-	World    World
+	World    resources.World
 	Noop     bool
 	NoWatch  bool
 	// NOTE: we can add more fields here if needed by GAPI endpoints
--- a/lib/cli.go
+++ b/lib/cli.go
@@ -84,6 +84,7 @@ func run(c *cli.Context) error {

 	obj.NoWatch = c.Bool("no-watch")
 	obj.Noop = c.Bool("noop")
+	obj.Sema = c.Int("sema")
 	obj.Graphviz = c.String("graphviz")
 	obj.GraphvizFilter = c.String("graphviz-filter")
 	obj.ConvergedTimeout = c.Int("converged-timeout")
@@ -228,6 +229,11 @@ func CLI(program, version string, flags Flags) error {
 					Name:  "noop",
 					Usage: "globally force all resources into no-op mode",
 				},
+				cli.IntFlag{
+					Name:  "sema",
+					Value: -1,
+					Usage: "globally add a semaphore to all resources with this lock count",
+				},
 				cli.StringFlag{
 					Name:  "graphviz, g",
 					Value: "",
@@ -235,7 +241,7 @@ func CLI(program, version string, flags Flags) error {
 				},
 				cli.StringFlag{
 					Name:  "graphviz-filter, gf",
-					Value: "dot", // directed graph default
+					Value: "",
 					Usage: "graphviz filter to use",
 				},
 				cli.IntFlag{
--- a/lib/main.go
+++ b/lib/main.go
@@ -67,6 +67,7 @@ type Main struct {

 	NoWatch          bool   // do not update graph on watched graph definition file changes
 	Noop             bool   // globally force all resources into no-op mode
+	Sema             int    // add a semaphore with this lock count to each resource
 	Graphviz         string // output file for graphviz data
 	GraphvizFilter   string // graphviz filter to use
 	ConvergedTimeout int    // exit after approximately this many seconds in a converged state; -1 to disable
@@ -348,15 +349,17 @@ func (obj *Main) Run() error {
 		converger.SetStateFn(convergerStateFn)
 	}

+	// implementation of the World API (alternates can be substituted in)
+	world := &etcd.World{
+		Hostname: hostname,
+		EmbdEtcd: EmbdEtcd,
+	}
+
 	var gapiChan chan error // stream events are nil errors
 	if obj.GAPI != nil {
 		data := gapi.Data{
 			Hostname: hostname,
-			// NOTE: alternate implementations can be substituted in
-			World: &etcd.World{
-				Hostname: hostname,
-				EmbdEtcd: EmbdEtcd,
-			},
+			World:    world,
 			Noop:     obj.Noop,
 			NoWatch:  obj.NoWatch,
 		}
@@ -437,17 +440,25 @@ func (obj *Main) Run() error {
 			newGraph.Flags = pgraph.Flags{Debug: obj.Flags.Debug}
 			// pass in the information we need
 			newGraph.AssociateData(&resources.Data{
+				Hostname:   hostname,
 				Converger:  converger,
 				Prometheus: prom,
+				World:      world,
 				Prefix:     pgraphPrefix,
 				Debug:      obj.Flags.Debug,
 			})

+			for _, m := range newGraph.GraphMetas() {
 				// apply the global noop parameter if requested
 				if obj.Noop {
-				for _, m := range newGraph.GraphMetas() {
 					m.Noop = obj.Noop
 				}
+
+				// append the semaphore to each resource
+				if obj.Sema > 0 { // NOTE: size == 0 would block
+					// a semaphore with an empty id is valid
+					m.Sema = append(m.Sema, fmt.Sprintf(":%d", obj.Sema))
+				}
 			}

 			// FIXME: make sure we "UnGroup()" any semi-destructive
@@ -472,13 +483,10 @@ func (obj *Main) Run() error {
 			// 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 != "" {
-				if err := G.ExecGraphviz(obj.GraphvizFilter, obj.Graphviz); err != nil {
-					log.Printf("Graphviz: %v", err)
-				} else {
-					log.Printf("Graphviz: Successfully generated graph!")
-				}
+			// Call this here because at this point the graph does not
+			// know anything about the prometheus instance.
+			if err := prom.UpdatePgraphStartTime(); err != nil {
+				log.Printf("Main: Prometheus.UpdatePgraphStartTime() errored: %v", err)
 			}
 			// G.Start(...) needs to be synchronous or wait,
 			// because if half of the nodes are started and
@@ -487,6 +495,19 @@ func (obj *Main) Run() error {
 			// even got going, thus causing nil pointer errors
 			G.Start(first)    // sync
 			converger.Start() // after G.Start()
+
+			log.Printf("Graph: %v", G) // show graph
+			if obj.Graphviz != "" {
+				filter := obj.GraphvizFilter
+				if filter == "" {
+					filter = "dot" // directed graph default
+				}
+				if err := G.ExecGraphviz(filter, obj.Graphviz, hostname); err != nil {
+					log.Printf("Graphviz: %v", err)
+				} else {
+					log.Printf("Graphviz: Successfully generated graph!")
+				}
+			}
 			first = false
 		}
 	}()
--- a/misc/make-deps.sh
+++ b/misc/make-deps.sh
@@ -15,13 +15,14 @@ YUM=`which yum 2>/dev/null`
 DNF=`which dnf 2>/dev/null`
 APT=`which apt-get 2>/dev/null`
 BREW=`which brew 2>/dev/null`
+PACMAN=`which pacman 2>/dev/null`

 # if DNF is available use it
 if [ -x "$DNF" ]; then
 	YUM=$DNF
 fi

-if [ -z "$YUM" -a -z "$APT" -a -z "$BREW" ]; then
+if [ -z "$YUM" -a -z "$APT" -a -z "$BREW" -a -z "$PACMAN" ]; then
 	echo "The package managers can't be found."
 	exit 1
 fi
@@ -41,6 +42,10 @@ if [ ! -z "$BREW" ]; then
 	$BREW install libvirt || true
 fi

+if [ ! -z "$PACMAN" ]; then
+	$sudo_command $PACMAN -S --noconfirm libvirt augeas libpcap
+fi
+
 if [ $travis -eq 0 ]; then
 	if [ ! -z "$YUM" ]; then
 		# some go dependencies are stored in mercurial
@@ -54,6 +59,9 @@ if [ $travis -eq 0 ]; then
 		$sudo_command $APT install -y golang-golang-x-tools || true
 		$sudo_command $APT install -y golang-go.tools || true
 	fi
+	if [ ! -z "$PACMAN" ]; then
+		$sudo_command $PACMAN -S --noconfirm go
+	fi
 fi

 # if golang is too old, we don't want to fail with an obscure error later
--- a/pgraph/actions.go
+++ b/pgraph/actions.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"log"
 	"math"
+	"strings"
 	"sync"
 	"time"

@@ -164,8 +165,30 @@ func (g *Graph) Process(v *Vertex) error {
 	if g.Flags.Debug {
 		log.Printf("%s[%s]: Process()", obj.Kind(), obj.GetName())
 	}
+	// FIXME: should these SetState methods be here or after the sema code?
 	defer obj.SetState(resources.ResStateNil) // reset state when finished
 	obj.SetState(resources.ResStateProcess)
+
+	// semaphores!
+	// These shouldn't ever block an exit, since the graph should eventually
+	// converge causing their them to unlock. More interestingly, since they
+	// run in a DAG alphabetically, there is no way to permanently deadlock,
+	// assuming that resources individually don't ever block from finishing!
+	// The exception is that semaphores with a zero count will always block!
+	// TODO: Add a close mechanism to close/unblock zero count semaphores...
+	semas := obj.Meta().Sema
+	if g.Flags.Debug && len(semas) > 0 {
+		log.Printf("%s[%s]: Sema: P(%s)", obj.Kind(), obj.GetName(), strings.Join(semas, ", "))
+	}
+	if err := g.SemaLock(semas); err != nil { // lock
+		// NOTE: in practice, this might not ever be truly necessary...
+		return fmt.Errorf("shutdown of semaphores")
+	}
+	defer g.SemaUnlock(semas) // unlock
+	if g.Flags.Debug && len(semas) > 0 {
+		defer log.Printf("%s[%s]: Sema: V(%s)", obj.Kind(), obj.GetName(), strings.Join(semas, ", "))
+	}
+
 	var ok = true
 	var applied = false // did we run an apply?
 	// is it okay to run dependency wise right now?
@@ -224,12 +247,10 @@ func (g *Graph) Process(v *Vertex) error {
 		// if this fails, don't UpdateTimestamp()
 		checkOK, err = obj.CheckApply(!noop)

-		if obj.Prometheus() != nil {
 		if promErr := obj.Prometheus().UpdateCheckApplyTotal(obj.Kind(), !noop, !checkOK, err != nil); promErr != nil {
 			// TODO: how to error correctly
 			log.Printf("%s[%s]: Prometheus.UpdateCheckApplyTotal() errored: %v", v.Kind(), v.GetName(), err)
 		}
-		}
 		// TODO: Can the `Poll` converged timeout tracking be a
 		// more general method for all converged timeouts? this
 		// would simplify the resources by removing boilerplate
@@ -307,6 +328,7 @@ func (obj *SentinelErr) Error() string {
 }

 // innerWorker is the CheckApply runner that reads from processChan.
+// TODO: would it be better if this was a method on BaseRes that took in *Graph?
 func (g *Graph) innerWorker(v *Vertex) {
 	obj := v.Res
 	running := false
@@ -344,6 +366,7 @@ Loop:
 					log.Printf("%s[%s]: Skipped event!", v.Kind(), v.GetName())
 				}
 				ev.ACK() // ready for next message
+				v.Res.QuiesceGroup().Done()
 				continue
 			}

@@ -355,6 +378,7 @@ Loop:
 				}
 				playback = true
 				ev.ACK() // ready for next message
+				v.Res.QuiesceGroup().Done()
 				continue
 			}

@@ -363,6 +387,7 @@ Loop:
 				e := fmt.Errorf("%s[%s]: Permanently limited (rate != Inf, burst: 0)", v.Kind(), v.GetName())
 				v.SendEvent(event.EventExit, &SentinelErr{e})
 				ev.ACK() // ready for next message
+				v.Res.QuiesceGroup().Done()
 				continue
 			}

@@ -382,6 +407,7 @@ Loop:
 					timer.Reset(d)
 					waiting = true // waiting for retry timer
 					ev.ACK()
+					v.Res.QuiesceGroup().Done()
 					continue
 				} // otherwise, we run directly!
 			}
@@ -398,6 +424,7 @@ Loop:
 					if retry == 0 {
 						// wrap the error in the sentinel
 						v.SendEvent(event.EventExit, &SentinelErr{e})
+						v.Res.QuiesceGroup().Done()
 						return
 					}
 					if retry > 0 { // don't decrement the -1
@@ -407,6 +434,8 @@ Loop:
 					// start the timer...
 					timer.Reset(delay)
 					waiting = true // waiting for retry timer
+					// don't v.Res.QuiesceGroup().Done() b/c
+					// the timer is running and it can exit!
 					return
 				}
 				retry = v.Meta().Retry // reset on success
@@ -436,15 +465,23 @@ Loop:
 			done = make(chan struct{}) // reset
 			// re-send this event, to trigger a CheckApply()
 			if playback {
-				playback = false
 				// this lock avoids us sending to
 				// channel after we've closed it!
 				// TODO: can this experience indefinite postponement ?
 				// see: https://github.com/golang/go/issues/11506
-				go obj.Event() // replay a new event
+				// pause or exit is in process if not quiescing!
+				if !v.Res.IsQuiescing() {
+					playback = false
+					v.Res.QuiesceGroup().Add(1) // lock around it, b/c still running...
+					go func() {
+						obj.Event() // replay a new event
+						v.Res.QuiesceGroup().Done()
+					}()
+				}
 			}
 			running = false
 			pcuid.SetConverged(true) // "unblock" Process
+			v.Res.QuiesceGroup().Done()

 		case <-wcuid.ConvergedTimer():
 			wcuid.SetConverged(true) // converged!
@@ -615,9 +652,16 @@ func (g *Graph) Start(first bool) { // start or continue
 	log.Printf("State: %v -> %v", g.setState(graphStateStarting), g.getState())
 	defer log.Printf("State: %v -> %v", g.setState(graphStateStarted), g.getState())
 	t, _ := g.TopologicalSort()
-	// TODO: only calculate indegree if `first` is true to save resources
 	indegree := g.InDegree() // compute all of the indegree's
-	for _, v := range Reverse(t) {
+	reversed := Reverse(t)
+	for _, v := range reversed { // run the Setup() for everyone first
+		if !v.Res.IsWorking() { // if Worker() is not running...
+			v.Res.Setup() // initialize some vars in the resource
+		}
+	}
+
+	// run through the topological reverse, and start or unpause each vertex
+	for _, v := range reversed {
 		// selective poke: here we reduce the number of initial pokes
 		// to the minimum required to activate every vertex in the
 		// graph, either by direct action, or by getting poked by a
@@ -632,10 +676,17 @@ func (g *Graph) Start(first bool) { // start or continue
 		// and not just selectively the subset with no indegree.

 		// let the startup code know to poke or not
-		v.Res.Starter((!first) || indegree[v] == 0)
+		// this triggers a CheckApply AFTER Watch is Running()
+		// We *don't* need to also do this to new nodes or nodes that
+		// are about to get unpaused, because they'll get poked by one
+		// of the indegree == 0 vertices, and an important aspect of the
+		// Process() function is that even if the state is correct, it
+		// will pass through the Poke so that it flows through the DAG.
+		v.Res.Starter(indegree[v] == 0)

+		var unpause = true
 		if !v.Res.IsWorking() { // if Worker() is not running...
-			v.Res.Setup()
+			unpause = false // doesn't need unpausing on first start
 			g.wg.Add(1)
 			// must pass in value to avoid races...
 			// see: https://ttboj.wordpress.com/2015/07/27/golang-parallelism-issues-causing-too-many-open-files-error/
@@ -660,7 +711,7 @@ func (g *Graph) Start(first bool) { // start or continue
 			// if the resource Init() fails, we don't hang!
 		}

-		if !first { // unpause!
+		if unpause { // unpause (if needed)
 			v.Res.SendEvent(event.EventStart, nil) // sync!
 		}
 	}
--- a/pgraph/autogroup_test.go
+++ b/pgraph/autogroup_test.go
@@ -0,0 +1,486 @@
+// Mgmt
+// Copyright (C) 2013-2017+ 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 pgraph
+
+import (
+	"testing"
+)
+
+// all of the following test cases are laid out with the following semantics:
+// * vertices which start with the same single letter are considered "like"
+// * "like" elements should be merged
+// * vertices can have any integer after their single letter "family" type
+// * grouped vertices should have a name with a comma separated list of names
+// * edges follow the same conventions about grouping
+
+// empty graph
+func TestPgraphGrouping1(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	g2 := NewGraph("g2") // expected result
+	runGraphCmp(t, g1, g2)
+}
+
+// single vertex
+func TestPgraphGrouping2(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{                    // grouping to limit variable scope
+		a1 := NewVertex(NewNoopResTest("a1"))
+		g1.AddVertex(a1)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		g2.AddVertex(a1)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+// two vertices
+func TestPgraphGrouping3(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		g1.AddVertex(a1, b1)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		g2.AddVertex(a1, b1)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+// two vertices merge
+func TestPgraphGrouping4(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		a2 := NewVertex(NewNoopResTest("a2"))
+		g1.AddVertex(a1, a2)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a := NewVertex(NewNoopResTest("a1,a2"))
+		g2.AddVertex(a)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+// three vertices merge
+func TestPgraphGrouping5(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		a2 := NewVertex(NewNoopResTest("a2"))
+		a3 := NewVertex(NewNoopResTest("a3"))
+		g1.AddVertex(a1, a2, a3)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a := NewVertex(NewNoopResTest("a1,a2,a3"))
+		g2.AddVertex(a)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+// three vertices, two merge
+func TestPgraphGrouping6(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		a2 := NewVertex(NewNoopResTest("a2"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		g1.AddVertex(a1, a2, b1)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a := NewVertex(NewNoopResTest("a1,a2"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		g2.AddVertex(a, b1)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+// four vertices, three merge
+func TestPgraphGrouping7(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		a2 := NewVertex(NewNoopResTest("a2"))
+		a3 := NewVertex(NewNoopResTest("a3"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		g1.AddVertex(a1, a2, a3, b1)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a := NewVertex(NewNoopResTest("a1,a2,a3"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		g2.AddVertex(a, b1)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+// four vertices, two&two merge
+func TestPgraphGrouping8(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		a2 := NewVertex(NewNoopResTest("a2"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		b2 := NewVertex(NewNoopResTest("b2"))
+		g1.AddVertex(a1, a2, b1, b2)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a := NewVertex(NewNoopResTest("a1,a2"))
+		b := NewVertex(NewNoopResTest("b1,b2"))
+		g2.AddVertex(a, b)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+// five vertices, two&three merge
+func TestPgraphGrouping9(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		a2 := NewVertex(NewNoopResTest("a2"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		b2 := NewVertex(NewNoopResTest("b2"))
+		b3 := NewVertex(NewNoopResTest("b3"))
+		g1.AddVertex(a1, a2, b1, b2, b3)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a := NewVertex(NewNoopResTest("a1,a2"))
+		b := NewVertex(NewNoopResTest("b1,b2,b3"))
+		g2.AddVertex(a, b)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+// three unique vertices
+func TestPgraphGrouping10(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		c1 := NewVertex(NewNoopResTest("c1"))
+		g1.AddVertex(a1, b1, c1)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		c1 := NewVertex(NewNoopResTest("c1"))
+		g2.AddVertex(a1, b1, c1)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+// three unique vertices, two merge
+func TestPgraphGrouping11(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		b2 := NewVertex(NewNoopResTest("b2"))
+		c1 := NewVertex(NewNoopResTest("c1"))
+		g1.AddVertex(a1, b1, b2, c1)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		b := NewVertex(NewNoopResTest("b1,b2"))
+		c1 := NewVertex(NewNoopResTest("c1"))
+		g2.AddVertex(a1, b, c1)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+// simple merge 1
+// a1   a2         a1,a2
+//   \ /     >>>     |     (arrows point downwards)
+//    b              b
+func TestPgraphGrouping12(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		a2 := NewVertex(NewNoopResTest("a2"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		e1 := NewEdge("e1")
+		e2 := NewEdge("e2")
+		g1.AddEdge(a1, b1, e1)
+		g1.AddEdge(a2, b1, e2)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a := NewVertex(NewNoopResTest("a1,a2"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		e := NewEdge("e1,e2")
+		g2.AddEdge(a, b1, e)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+// simple merge 2
+//    b              b
+//   / \     >>>     |     (arrows point downwards)
+// a1   a2         a1,a2
+func TestPgraphGrouping13(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		a2 := NewVertex(NewNoopResTest("a2"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		e1 := NewEdge("e1")
+		e2 := NewEdge("e2")
+		g1.AddEdge(b1, a1, e1)
+		g1.AddEdge(b1, a2, e2)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a := NewVertex(NewNoopResTest("a1,a2"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		e := NewEdge("e1,e2")
+		g2.AddEdge(b1, a, e)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+// triple merge
+// a1 a2  a3         a1,a2,a3
+//   \ | /     >>>       |      (arrows point downwards)
+//     b                 b
+func TestPgraphGrouping14(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		a2 := NewVertex(NewNoopResTest("a2"))
+		a3 := NewVertex(NewNoopResTest("a3"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		e1 := NewEdge("e1")
+		e2 := NewEdge("e2")
+		e3 := NewEdge("e3")
+		g1.AddEdge(a1, b1, e1)
+		g1.AddEdge(a2, b1, e2)
+		g1.AddEdge(a3, b1, e3)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a := NewVertex(NewNoopResTest("a1,a2,a3"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		e := NewEdge("e1,e2,e3")
+		g2.AddEdge(a, b1, e)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+// chain merge
+//    a1             a1
+//   /  \             |
+// b1    b2   >>>   b1,b2   (arrows point downwards)
+//   \  /             |
+//    c1             c1
+func TestPgraphGrouping15(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		b2 := NewVertex(NewNoopResTest("b2"))
+		c1 := NewVertex(NewNoopResTest("c1"))
+		e1 := NewEdge("e1")
+		e2 := NewEdge("e2")
+		e3 := NewEdge("e3")
+		e4 := NewEdge("e4")
+		g1.AddEdge(a1, b1, e1)
+		g1.AddEdge(a1, b2, e2)
+		g1.AddEdge(b1, c1, e3)
+		g1.AddEdge(b2, c1, e4)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		b := NewVertex(NewNoopResTest("b1,b2"))
+		c1 := NewVertex(NewNoopResTest("c1"))
+		e1 := NewEdge("e1,e2")
+		e2 := NewEdge("e3,e4")
+		g2.AddEdge(a1, b, e1)
+		g2.AddEdge(b, c1, e2)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+// re-attach 1 (outer)
+// technically the second possibility is valid too, depending on which order we
+// merge edges in, and if we don't filter out any unnecessary edges afterwards!
+// a1    a2         a1,a2        a1,a2
+//  |   /             |            |  \
+// b1  /      >>>    b1     OR    b1  /   (arrows point downwards)
+//  | /               |            | /
+// c1                c1           c1
+func TestPgraphGrouping16(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		a2 := NewVertex(NewNoopResTest("a2"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		c1 := NewVertex(NewNoopResTest("c1"))
+		e1 := NewEdge("e1")
+		e2 := NewEdge("e2")
+		e3 := NewEdge("e3")
+		g1.AddEdge(a1, b1, e1)
+		g1.AddEdge(b1, c1, e2)
+		g1.AddEdge(a2, c1, e3)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a := NewVertex(NewNoopResTest("a1,a2"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		c1 := NewVertex(NewNoopResTest("c1"))
+		e1 := NewEdge("e1,e3")
+		e2 := NewEdge("e2,e3") // e3 gets "merged through" to BOTH edges!
+		g2.AddEdge(a, b1, e1)
+		g2.AddEdge(b1, c1, e2)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+// re-attach 2 (inner)
+// a1    b2          a1
+//  |   /             |
+// b1  /      >>>   b1,b2   (arrows point downwards)
+//  | /               |
+// c1                c1
+func TestPgraphGrouping17(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		b2 := NewVertex(NewNoopResTest("b2"))
+		c1 := NewVertex(NewNoopResTest("c1"))
+		e1 := NewEdge("e1")
+		e2 := NewEdge("e2")
+		e3 := NewEdge("e3")
+		g1.AddEdge(a1, b1, e1)
+		g1.AddEdge(b1, c1, e2)
+		g1.AddEdge(b2, c1, e3)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		b := NewVertex(NewNoopResTest("b1,b2"))
+		c1 := NewVertex(NewNoopResTest("c1"))
+		e1 := NewEdge("e1")
+		e2 := NewEdge("e2,e3")
+		g2.AddEdge(a1, b, e1)
+		g2.AddEdge(b, c1, e2)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+// re-attach 3 (double)
+// similar to "re-attach 1", technically there is a second possibility for this
+// a2   a1    b2         a1,a2
+//   \   |   /             |
+//    \ b1  /      >>>   b1,b2   (arrows point downwards)
+//     \ | /               |
+//      c1                c1
+func TestPgraphGrouping18(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		a2 := NewVertex(NewNoopResTest("a2"))
+		b1 := NewVertex(NewNoopResTest("b1"))
+		b2 := NewVertex(NewNoopResTest("b2"))
+		c1 := NewVertex(NewNoopResTest("c1"))
+		e1 := NewEdge("e1")
+		e2 := NewEdge("e2")
+		e3 := NewEdge("e3")
+		e4 := NewEdge("e4")
+		g1.AddEdge(a1, b1, e1)
+		g1.AddEdge(b1, c1, e2)
+		g1.AddEdge(a2, c1, e3)
+		g1.AddEdge(b2, c1, e4)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a := NewVertex(NewNoopResTest("a1,a2"))
+		b := NewVertex(NewNoopResTest("b1,b2"))
+		c1 := NewVertex(NewNoopResTest("c1"))
+		e1 := NewEdge("e1,e3")
+		e2 := NewEdge("e2,e3,e4") // e3 gets "merged through" to BOTH edges!
+		g2.AddEdge(a, b, e1)
+		g2.AddEdge(b, c1, e2)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+// connected merge 0, (no change!)
+// a1            a1
+//   \     >>>     \     (arrows point downwards)
+//    a2            a2
+func TestPgraphGroupingConnected0(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		a2 := NewVertex(NewNoopResTest("a2"))
+		e1 := NewEdge("e1")
+		g1.AddEdge(a1, a2, e1)
+	}
+	g2 := NewGraph("g2") // expected result ?
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		a2 := NewVertex(NewNoopResTest("a2"))
+		e1 := NewEdge("e1")
+		g2.AddEdge(a1, a2, e1)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+// connected merge 1, (no change!)
+// a1              a1
+//   \               \
+//    b      >>>      b      (arrows point downwards)
+//     \               \
+//      a2              a2
+func TestPgraphGroupingConnected1(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		b := NewVertex(NewNoopResTest("b"))
+		a2 := NewVertex(NewNoopResTest("a2"))
+		e1 := NewEdge("e1")
+		e2 := NewEdge("e2")
+		g1.AddEdge(a1, b, e1)
+		g1.AddEdge(b, a2, e2)
+	}
+	g2 := NewGraph("g2") // expected result ?
+	{
+		a1 := NewVertex(NewNoopResTest("a1"))
+		b := NewVertex(NewNoopResTest("b"))
+		a2 := NewVertex(NewNoopResTest("a2"))
+		e1 := NewEdge("e1")
+		e2 := NewEdge("e2")
+		g2.AddEdge(a1, b, e1)
+		g2.AddEdge(b, a2, e2)
+	}
+	runGraphCmp(t, g1, g2)
+}
--- a/pgraph/graphviz.go
+++ b/pgraph/graphviz.go
@@ -15,7 +15,7 @@
 // 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 pgraph
+package pgraph // TODO: this should be a subpackage

 import (
 	"fmt"
@@ -46,14 +46,14 @@ func (g *Graph) Graphviz() (out string) {
 	//out += "\tnode [shape=box];\n"
 	str := ""
 	for i := range g.Adjacency { // reverse paths
-		out += fmt.Sprintf("\t%s [label=\"%s[%s]\"];\n", i.GetName(), i.Kind(), i.GetName())
+		out += fmt.Sprintf("\t\"%s\" [label=\"%s[%s]\"];\n", i.GetName(), i.Kind(), i.GetName())
 		for j := range g.Adjacency[i] {
 			k := g.Adjacency[i][j]
 			// use str for clearer output ordering
 			if k.Notify {
-				str += fmt.Sprintf("\t%s -> %s [label=%s,style=bold];\n", i.GetName(), j.GetName(), k.Name)
+				str += fmt.Sprintf("\t\"%s\" -> \"%s\" [label=\"%s\",style=bold];\n", i.GetName(), j.GetName(), k.Name)
 			} else {
-				str += fmt.Sprintf("\t%s -> %s [label=%s];\n", i.GetName(), j.GetName(), k.Name)
+				str += fmt.Sprintf("\t\"%s\" -> \"%s\" [label=\"%s\"];\n", i.GetName(), j.GetName(), k.Name)
 			}
 		}
 	}
@@ -64,7 +64,7 @@ func (g *Graph) Graphviz() (out string) {

 // ExecGraphviz writes out the graphviz data and runs the correct graphviz
 // filter command.
-func (g *Graph) ExecGraphviz(program, filename string) error {
+func (g *Graph) ExecGraphviz(program, filename, hostname string) error {

 	switch program {
 	case "dot", "neato", "twopi", "circo", "fdp":
@@ -76,6 +76,10 @@ func (g *Graph) ExecGraphviz(program, filename string) error {
 		return fmt.Errorf("no filename given")
 	}

+	if hostname != "" {
+		filename = fmt.Sprintf("%s@%s", filename, hostname)
+	}
+
 	// run as a normal user if possible when run with sudo
 	uid, err1 := strconv.Atoi(os.Getenv("SUDO_UID"))
 	gid, err2 := strconv.Atoi(os.Getenv("SUDO_GID"))
--- a/pgraph/pgraph.go
+++ b/pgraph/pgraph.go
@@ -24,7 +24,9 @@ import (
 	"sync"

 	"github.com/purpleidea/mgmt/event"
+	"github.com/purpleidea/mgmt/prometheus"
 	"github.com/purpleidea/mgmt/resources"
+	"github.com/purpleidea/mgmt/util/semaphore"

 	errwrap "github.com/pkg/errors"
 )
@@ -58,6 +60,9 @@ type Graph struct {
 	state     graphState
 	mutex     *sync.Mutex // used when modifying graph State variable
 	wg        *sync.WaitGroup
+	semas     map[string]*semaphore.Semaphore
+
+	prometheus *prometheus.Prometheus // the prometheus instance
 }

 // Vertex is the primary vertex struct in this library.
@@ -83,6 +88,7 @@ func NewGraph(name string) *Graph {
 		// ptr b/c: Mutex/WaitGroup must not be copied after first use
 		mutex: &sync.Mutex{},
 		wg:    &sync.WaitGroup{},
+		semas: make(map[string]*semaphore.Semaphore),
 	}
 }

@@ -119,6 +125,9 @@ func (g *Graph) Copy() *Graph {
 		state:     g.state,
 		mutex:     g.mutex,
 		wg:        g.wg,
+		semas:     g.semas,
+
+		prometheus: g.prometheus,
 	}
 	for k, v := range g.Adjacency {
 		newGraph.Adjacency[k] = v // copy
@@ -645,8 +654,11 @@ func (g *Graph) GraphMetas() []*resources.MetaParams {

 // AssociateData associates some data with the object in the graph in question.
 func (g *Graph) AssociateData(data *resources.Data) {
+	// prometheus needs to be associated to this graph as well
+	g.prometheus = data.Prometheus
+
 	for k := range g.Adjacency {
-		k.Res.AssociateData(data)
+		*k.Res.Data() = *data
 	}
 }

--- a/pgraph/pgraph_test.go
+++ b/pgraph/pgraph_test.go
@@ -718,7 +718,7 @@ Loop:
 				continue Loop
 			}
 		}
-		return fmt.Errorf("fraph g1, has no match in g2 for: %v", v1.GetName())
+		return fmt.Errorf("graph g1, has no match in g2 for: %v", v1.GetName())
 	}
 	// vertices (and groups) match :)

@@ -764,6 +764,18 @@ Loop:
 		}
 	}

+	// check meta parameters
+	for v1 := range g1.Adjacency { // for each vertex in g1
+		for v2 := range g2.Adjacency { // does it match in g2 ?
+			s1, s2 := v1.Meta().Sema, v2.Meta().Sema
+			sort.Strings(s1)
+			sort.Strings(s2)
+			if !reflect.DeepEqual(s1, s2) {
+				return fmt.Errorf("vertex %s and vertex %s have different semaphores", v1.GetName(), v2.GetName())
+			}
+		}
+	}
+
 	return nil // success!
 }

@@ -805,7 +817,11 @@ func (ag *testGrouper) edgeMerge(e1, e2 *Edge) *Edge {
 func (g *Graph) fullPrint() (str string) {
 	str += "\n"
 	for v := range g.Adjacency {
+		if semas := v.Meta().Sema; len(semas) > 0 {
+			str += fmt.Sprintf("* v: %v; sema: %v\n", v.GetName(), semas)
+		} else {
 			str += fmt.Sprintf("* v: %v\n", v.GetName())
+		}
 		// TODO: add explicit grouping data?
 	}
 	for v1 := range g.Adjacency {
@@ -831,470 +847,6 @@ func runGraphCmp(t *testing.T, g1, g2 *Graph) {
 	}
 }

-// all of the following test cases are laid out with the following semantics:
-// * vertices which start with the same single letter are considered "like"
-// * "like" elements should be merged
-// * vertices can have any integer after their single letter "family" type
-// * grouped vertices should have a name with a comma separated list of names
-// * edges follow the same conventions about grouping
-
-// empty graph
-func TestPgraphGrouping1(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	g2 := NewGraph("g2") // expected result
-	runGraphCmp(t, g1, g2)
-}
-
-// single vertex
-func TestPgraphGrouping2(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{                    // grouping to limit variable scope
-		a1 := NewVertex(NewNoopResTest("a1"))
-		g1.AddVertex(a1)
-	}
-	g2 := NewGraph("g2") // expected result
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		g2.AddVertex(a1)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
-// two vertices
-func TestPgraphGrouping3(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		g1.AddVertex(a1, b1)
-	}
-	g2 := NewGraph("g2") // expected result
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		g2.AddVertex(a1, b1)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
-// two vertices merge
-func TestPgraphGrouping4(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		a2 := NewVertex(NewNoopResTest("a2"))
-		g1.AddVertex(a1, a2)
-	}
-	g2 := NewGraph("g2") // expected result
-	{
-		a := NewVertex(NewNoopResTest("a1,a2"))
-		g2.AddVertex(a)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
-// three vertices merge
-func TestPgraphGrouping5(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		a2 := NewVertex(NewNoopResTest("a2"))
-		a3 := NewVertex(NewNoopResTest("a3"))
-		g1.AddVertex(a1, a2, a3)
-	}
-	g2 := NewGraph("g2") // expected result
-	{
-		a := NewVertex(NewNoopResTest("a1,a2,a3"))
-		g2.AddVertex(a)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
-// three vertices, two merge
-func TestPgraphGrouping6(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		a2 := NewVertex(NewNoopResTest("a2"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		g1.AddVertex(a1, a2, b1)
-	}
-	g2 := NewGraph("g2") // expected result
-	{
-		a := NewVertex(NewNoopResTest("a1,a2"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		g2.AddVertex(a, b1)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
-// four vertices, three merge
-func TestPgraphGrouping7(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		a2 := NewVertex(NewNoopResTest("a2"))
-		a3 := NewVertex(NewNoopResTest("a3"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		g1.AddVertex(a1, a2, a3, b1)
-	}
-	g2 := NewGraph("g2") // expected result
-	{
-		a := NewVertex(NewNoopResTest("a1,a2,a3"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		g2.AddVertex(a, b1)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
-// four vertices, two&two merge
-func TestPgraphGrouping8(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		a2 := NewVertex(NewNoopResTest("a2"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		b2 := NewVertex(NewNoopResTest("b2"))
-		g1.AddVertex(a1, a2, b1, b2)
-	}
-	g2 := NewGraph("g2") // expected result
-	{
-		a := NewVertex(NewNoopResTest("a1,a2"))
-		b := NewVertex(NewNoopResTest("b1,b2"))
-		g2.AddVertex(a, b)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
-// five vertices, two&three merge
-func TestPgraphGrouping9(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		a2 := NewVertex(NewNoopResTest("a2"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		b2 := NewVertex(NewNoopResTest("b2"))
-		b3 := NewVertex(NewNoopResTest("b3"))
-		g1.AddVertex(a1, a2, b1, b2, b3)
-	}
-	g2 := NewGraph("g2") // expected result
-	{
-		a := NewVertex(NewNoopResTest("a1,a2"))
-		b := NewVertex(NewNoopResTest("b1,b2,b3"))
-		g2.AddVertex(a, b)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
-// three unique vertices
-func TestPgraphGrouping10(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		c1 := NewVertex(NewNoopResTest("c1"))
-		g1.AddVertex(a1, b1, c1)
-	}
-	g2 := NewGraph("g2") // expected result
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		c1 := NewVertex(NewNoopResTest("c1"))
-		g2.AddVertex(a1, b1, c1)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
-// three unique vertices, two merge
-func TestPgraphGrouping11(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		b2 := NewVertex(NewNoopResTest("b2"))
-		c1 := NewVertex(NewNoopResTest("c1"))
-		g1.AddVertex(a1, b1, b2, c1)
-	}
-	g2 := NewGraph("g2") // expected result
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		b := NewVertex(NewNoopResTest("b1,b2"))
-		c1 := NewVertex(NewNoopResTest("c1"))
-		g2.AddVertex(a1, b, c1)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
-// simple merge 1
-// a1   a2         a1,a2
-//   \ /     >>>     |     (arrows point downwards)
-//    b              b
-func TestPgraphGrouping12(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		a2 := NewVertex(NewNoopResTest("a2"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		e1 := NewEdge("e1")
-		e2 := NewEdge("e2")
-		g1.AddEdge(a1, b1, e1)
-		g1.AddEdge(a2, b1, e2)
-	}
-	g2 := NewGraph("g2") // expected result
-	{
-		a := NewVertex(NewNoopResTest("a1,a2"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		e := NewEdge("e1,e2")
-		g2.AddEdge(a, b1, e)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
-// simple merge 2
-//    b              b
-//   / \     >>>     |     (arrows point downwards)
-// a1   a2         a1,a2
-func TestPgraphGrouping13(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		a2 := NewVertex(NewNoopResTest("a2"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		e1 := NewEdge("e1")
-		e2 := NewEdge("e2")
-		g1.AddEdge(b1, a1, e1)
-		g1.AddEdge(b1, a2, e2)
-	}
-	g2 := NewGraph("g2") // expected result
-	{
-		a := NewVertex(NewNoopResTest("a1,a2"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		e := NewEdge("e1,e2")
-		g2.AddEdge(b1, a, e)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
-// triple merge
-// a1 a2  a3         a1,a2,a3
-//   \ | /     >>>       |      (arrows point downwards)
-//     b                 b
-func TestPgraphGrouping14(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		a2 := NewVertex(NewNoopResTest("a2"))
-		a3 := NewVertex(NewNoopResTest("a3"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		e1 := NewEdge("e1")
-		e2 := NewEdge("e2")
-		e3 := NewEdge("e3")
-		g1.AddEdge(a1, b1, e1)
-		g1.AddEdge(a2, b1, e2)
-		g1.AddEdge(a3, b1, e3)
-	}
-	g2 := NewGraph("g2") // expected result
-	{
-		a := NewVertex(NewNoopResTest("a1,a2,a3"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		e := NewEdge("e1,e2,e3")
-		g2.AddEdge(a, b1, e)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
-// chain merge
-//    a1             a1
-//   /  \             |
-// b1    b2   >>>   b1,b2   (arrows point downwards)
-//   \  /             |
-//    c1             c1
-func TestPgraphGrouping15(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		b2 := NewVertex(NewNoopResTest("b2"))
-		c1 := NewVertex(NewNoopResTest("c1"))
-		e1 := NewEdge("e1")
-		e2 := NewEdge("e2")
-		e3 := NewEdge("e3")
-		e4 := NewEdge("e4")
-		g1.AddEdge(a1, b1, e1)
-		g1.AddEdge(a1, b2, e2)
-		g1.AddEdge(b1, c1, e3)
-		g1.AddEdge(b2, c1, e4)
-	}
-	g2 := NewGraph("g2") // expected result
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		b := NewVertex(NewNoopResTest("b1,b2"))
-		c1 := NewVertex(NewNoopResTest("c1"))
-		e1 := NewEdge("e1,e2")
-		e2 := NewEdge("e3,e4")
-		g2.AddEdge(a1, b, e1)
-		g2.AddEdge(b, c1, e2)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
-// re-attach 1 (outer)
-// technically the second possibility is valid too, depending on which order we
-// merge edges in, and if we don't filter out any unnecessary edges afterwards!
-// a1    a2         a1,a2        a1,a2
-//  |   /             |            |  \
-// b1  /      >>>    b1     OR    b1  /   (arrows point downwards)
-//  | /               |            | /
-// c1                c1           c1
-func TestPgraphGrouping16(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		a2 := NewVertex(NewNoopResTest("a2"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		c1 := NewVertex(NewNoopResTest("c1"))
-		e1 := NewEdge("e1")
-		e2 := NewEdge("e2")
-		e3 := NewEdge("e3")
-		g1.AddEdge(a1, b1, e1)
-		g1.AddEdge(b1, c1, e2)
-		g1.AddEdge(a2, c1, e3)
-	}
-	g2 := NewGraph("g2") // expected result
-	{
-		a := NewVertex(NewNoopResTest("a1,a2"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		c1 := NewVertex(NewNoopResTest("c1"))
-		e1 := NewEdge("e1,e3")
-		e2 := NewEdge("e2,e3") // e3 gets "merged through" to BOTH edges!
-		g2.AddEdge(a, b1, e1)
-		g2.AddEdge(b1, c1, e2)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
-// re-attach 2 (inner)
-// a1    b2          a1
-//  |   /             |
-// b1  /      >>>   b1,b2   (arrows point downwards)
-//  | /               |
-// c1                c1
-func TestPgraphGrouping17(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		b2 := NewVertex(NewNoopResTest("b2"))
-		c1 := NewVertex(NewNoopResTest("c1"))
-		e1 := NewEdge("e1")
-		e2 := NewEdge("e2")
-		e3 := NewEdge("e3")
-		g1.AddEdge(a1, b1, e1)
-		g1.AddEdge(b1, c1, e2)
-		g1.AddEdge(b2, c1, e3)
-	}
-	g2 := NewGraph("g2") // expected result
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		b := NewVertex(NewNoopResTest("b1,b2"))
-		c1 := NewVertex(NewNoopResTest("c1"))
-		e1 := NewEdge("e1")
-		e2 := NewEdge("e2,e3")
-		g2.AddEdge(a1, b, e1)
-		g2.AddEdge(b, c1, e2)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
-// re-attach 3 (double)
-// similar to "re-attach 1", technically there is a second possibility for this
-// a2   a1    b2         a1,a2
-//   \   |   /             |
-//    \ b1  /      >>>   b1,b2   (arrows point downwards)
-//     \ | /               |
-//      c1                c1
-func TestPgraphGrouping18(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		a2 := NewVertex(NewNoopResTest("a2"))
-		b1 := NewVertex(NewNoopResTest("b1"))
-		b2 := NewVertex(NewNoopResTest("b2"))
-		c1 := NewVertex(NewNoopResTest("c1"))
-		e1 := NewEdge("e1")
-		e2 := NewEdge("e2")
-		e3 := NewEdge("e3")
-		e4 := NewEdge("e4")
-		g1.AddEdge(a1, b1, e1)
-		g1.AddEdge(b1, c1, e2)
-		g1.AddEdge(a2, c1, e3)
-		g1.AddEdge(b2, c1, e4)
-	}
-	g2 := NewGraph("g2") // expected result
-	{
-		a := NewVertex(NewNoopResTest("a1,a2"))
-		b := NewVertex(NewNoopResTest("b1,b2"))
-		c1 := NewVertex(NewNoopResTest("c1"))
-		e1 := NewEdge("e1,e3")
-		e2 := NewEdge("e2,e3,e4") // e3 gets "merged through" to BOTH edges!
-		g2.AddEdge(a, b, e1)
-		g2.AddEdge(b, c1, e2)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
-// connected merge 0, (no change!)
-// a1            a1
-//   \     >>>     \     (arrows point downwards)
-//    a2            a2
-func TestPgraphGroupingConnected0(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		a2 := NewVertex(NewNoopResTest("a2"))
-		e1 := NewEdge("e1")
-		g1.AddEdge(a1, a2, e1)
-	}
-	g2 := NewGraph("g2") // expected result ?
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		a2 := NewVertex(NewNoopResTest("a2"))
-		e1 := NewEdge("e1")
-		g2.AddEdge(a1, a2, e1)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
-// connected merge 1, (no change!)
-// a1              a1
-//   \               \
-//    b      >>>      b      (arrows point downwards)
-//     \               \
-//      a2              a2
-func TestPgraphGroupingConnected1(t *testing.T) {
-	g1 := NewGraph("g1") // original graph
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		b := NewVertex(NewNoopResTest("b"))
-		a2 := NewVertex(NewNoopResTest("a2"))
-		e1 := NewEdge("e1")
-		e2 := NewEdge("e2")
-		g1.AddEdge(a1, b, e1)
-		g1.AddEdge(b, a2, e2)
-	}
-	g2 := NewGraph("g2") // expected result ?
-	{
-		a1 := NewVertex(NewNoopResTest("a1"))
-		b := NewVertex(NewNoopResTest("b"))
-		a2 := NewVertex(NewNoopResTest("a2"))
-		e1 := NewEdge("e1")
-		e2 := NewEdge("e2")
-		g2.AddEdge(a1, b, e1)
-		g2.AddEdge(b, a2, e2)
-	}
-	runGraphCmp(t, g1, g2)
-}
-
 func TestDurationAssumptions(t *testing.T) {
 	var d time.Duration
 	if (d == 0) != true {
--- a/pgraph/semaphore.go
+++ b/pgraph/semaphore.go
@@ -0,0 +1,78 @@
+// Mgmt
+// Copyright (C) 2013-2017+ 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 pgraph
+
+import (
+	"fmt"
+	"sort"
+	"strconv"
+	"strings"
+
+	"github.com/purpleidea/mgmt/util/semaphore"
+
+	multierr "github.com/hashicorp/go-multierror"
+)
+
+// SemaSep is the trailing separator to split the semaphore id from the size.
+const SemaSep = ":"
+
+// SemaLock acquires the list of semaphores in the graph.
+func (g *Graph) SemaLock(semas []string) error {
+	var reterr error
+	sort.Strings(semas) // very important to avoid deadlock in the dag!
+	for _, id := range semas {
+
+		size := 1 // default semaphore size
+		// valid id's include "some_id", "hello:42" and ":13"
+		if index := strings.LastIndex(id, SemaSep); index > -1 && (len(id)-index+len(SemaSep)) >= 1 {
+			// NOTE: we only allow size > 0 here!
+			if i, err := strconv.Atoi(id[index+len(SemaSep):]); err == nil && i > 0 {
+				size = i
+			}
+		}
+
+		sema, ok := g.semas[id] // lookup
+		if !ok {
+			g.semas[id] = semaphore.NewSemaphore(size)
+			sema = g.semas[id]
+		}
+
+		if err := sema.P(1); err != nil { // lock!
+			reterr = multierr.Append(reterr, err) // list of errors
+		}
+	}
+	return reterr
+}
+
+// SemaUnlock releases the list of semaphores in the graph.
+func (g *Graph) SemaUnlock(semas []string) error {
+	var reterr error
+	sort.Strings(semas) // unlock in the same order to remove partial locks
+	for _, id := range semas {
+		sema, ok := g.semas[id] // lookup
+		if !ok {
+			// programming error!
+			panic(fmt.Sprintf("graph: sema: %s does not exist", id))
+		}
+
+		if err := sema.V(1); err != nil { // unlock!
+			reterr = multierr.Append(reterr, err) // list of errors
+		}
+	}
+	return reterr
+}
--- a/pgraph/semaphore_test.go
+++ b/pgraph/semaphore_test.go
@@ -0,0 +1,93 @@
+// Mgmt
+// Copyright (C) 2013-2017+ 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 pgraph
+
+import (
+	"testing"
+
+	"github.com/purpleidea/mgmt/resources"
+)
+
+func NewNoopResTestSema(name string, semas []string) *NoopResTest {
+	obj := &NoopResTest{
+		NoopRes: resources.NoopRes{
+			BaseRes: resources.BaseRes{
+				Name: name,
+				MetaParams: resources.MetaParams{
+					AutoGroup: true, // always autogroup
+					Sema:      semas,
+				},
+			},
+		},
+	}
+	return obj
+}
+
+func TestPgraphSemaphoreGrouping1(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTestSema("a1", []string{"s:1"}))
+		a2 := NewVertex(NewNoopResTestSema("a2", []string{"s:2"}))
+		a3 := NewVertex(NewNoopResTestSema("a3", []string{"s:3"}))
+		g1.AddVertex(a1)
+		g1.AddVertex(a2)
+		g1.AddVertex(a3)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a123 := NewVertex(NewNoopResTestSema("a1,a2,a3", []string{"s:1", "s:2", "s:3"}))
+		g2.AddVertex(a123)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+func TestPgraphSemaphoreGrouping2(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTestSema("a1", []string{"s:10", "s:11"}))
+		a2 := NewVertex(NewNoopResTestSema("a2", []string{"s:2"}))
+		a3 := NewVertex(NewNoopResTestSema("a3", []string{"s:3"}))
+		g1.AddVertex(a1)
+		g1.AddVertex(a2)
+		g1.AddVertex(a3)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a123 := NewVertex(NewNoopResTestSema("a1,a2,a3", []string{"s:10", "s:11", "s:2", "s:3"}))
+		g2.AddVertex(a123)
+	}
+	runGraphCmp(t, g1, g2)
+}
+
+func TestPgraphSemaphoreGrouping3(t *testing.T) {
+	g1 := NewGraph("g1") // original graph
+	{
+		a1 := NewVertex(NewNoopResTestSema("a1", []string{"s:1", "s:2"}))
+		a2 := NewVertex(NewNoopResTestSema("a2", []string{"s:2"}))
+		a3 := NewVertex(NewNoopResTestSema("a3", []string{"s:3"}))
+		g1.AddVertex(a1)
+		g1.AddVertex(a2)
+		g1.AddVertex(a3)
+	}
+	g2 := NewGraph("g2") // expected result
+	{
+		a123 := NewVertex(NewNoopResTestSema("a1,a2,a3", []string{"s:1", "s:2", "s:3"}))
+		g2.AddVertex(a123)
+	}
+	runGraphCmp(t, g1, g2)
+}
--- a/prometheus/prometheus.go
+++ b/prometheus/prometheus.go
@@ -37,6 +37,7 @@ type Prometheus struct {
 	Listen string // the listen specification for the net/http server

 	checkApplyTotal        *prometheus.CounterVec // total of CheckApplies that have been triggered
+	pgraphStartTimeSeconds prometheus.Gauge       // process start time in seconds since unix epoch

 }

@@ -59,6 +60,14 @@ func (obj *Prometheus) Init() error {
 	)
 	prometheus.MustRegister(obj.checkApplyTotal)

+	obj.pgraphStartTimeSeconds = prometheus.NewGauge(
+		prometheus.GaugeOpts{
+			Name: "mgmt_graph_start_time_seconds",
+			Help: "Start time of the current graph since unix epoch in seconds.",
+		},
+	)
+	prometheus.MustRegister(obj.pgraphStartTimeSeconds)
+
 	return nil
 }

@@ -80,8 +89,21 @@ func (obj *Prometheus) Stop() error {
 // UpdateCheckApplyTotal refreshes the failing gauge by parsing the internal
 // state map.
 func (obj *Prometheus) UpdateCheckApplyTotal(kind string, apply, eventful, errorful bool) error {
+	if obj == nil {
+		return nil // happens when mgmt is launched without --prometheus
+	}
 	labels := prometheus.Labels{"kind": kind, "apply": strconv.FormatBool(apply), "eventful": strconv.FormatBool(eventful), "errorful": strconv.FormatBool(errorful)}
 	metric := obj.checkApplyTotal.With(labels)
 	metric.Inc()
 	return nil
 }
+
+// UpdatePgraphStartTime updates the mgmt_graph_start_time_seconds metric
+// to the current timestamp.
+func (obj *Prometheus) UpdatePgraphStartTime() error {
+	if obj == nil {
+		return nil // happens when mgmt is launched without --prometheus
+	}
+	obj.pgraphStartTimeSeconds.SetToCurrentTime()
+	return nil
+}
--- a/remote/remote.go
+++ b/remote/remote.go
@@ -63,6 +63,7 @@ import (

 	cv "github.com/purpleidea/mgmt/converger"
 	"github.com/purpleidea/mgmt/util"
+	"github.com/purpleidea/mgmt/util/semaphore"
 	"github.com/purpleidea/mgmt/yamlgraph"

 	multierr "github.com/hashicorp/go-multierror"
@@ -703,7 +704,7 @@ type Remotes struct {
 	sshmap             map[string]*SSH      // map to each SSH struct with the remote as the key
 	exiting            bool                 // flag to let us know if we're exiting
 	exitChan           chan struct{}        // closes when we should exit
-	semaphore          Semaphore         // counting semaphore to limit concurrent connections
+	semaphore          *semaphore.Semaphore // counting semaphore to limit concurrent connections
 	hostnames          []string             // list of hostnames we've seen so far
 	cuid               cv.UID               // convergerUID for the remote itself
 	cuids              map[string]cv.UID    // map to each SSH struct with the remote as the key
@@ -730,7 +731,7 @@ func NewRemotes(clientURLs, remoteURLs []string, noop bool, remotes []string, fi
 		convergerCb:  convergerCb,
 		sshmap:       make(map[string]*SSH),
 		exitChan:     make(chan struct{}),
-		semaphore:    NewSemaphore(int(cConns)),
+		semaphore:    semaphore.NewSemaphore(int(cConns)),
 		hostnames:    make([]string, len(remotes)),
 		cuids:        make(map[string]cv.UID),
 		flags:        flags,
@@ -1078,29 +1079,6 @@ func cleanURL(s string) string {
 	return u.Host
 }

-// Semaphore is a counting semaphore.
-type Semaphore chan struct{}
-
-// NewSemaphore creates a new semaphore.
-func NewSemaphore(size int) Semaphore {
-	return make(Semaphore, size)
-}
-
-// P acquires n resources.
-func (s Semaphore) P(n int) {
-	e := struct{}{}
-	for i := 0; i < n; i++ {
-		s <- e // acquire one
-	}
-}
-
-// V releases n resources.
-func (s Semaphore) V(n int) {
-	for i := 0; i < n; i++ {
-		<-s // release one
-	}
-}
-
 // combinedWriter mimics what the ssh.CombinedOutput command does.
 type combinedWriter struct {
 	b  bytes.Buffer
--- a/resources/augeas.go
+++ b/resources/augeas.go
@@ -47,7 +47,7 @@ func init() {
 type AugeasRes struct {
 	BaseRes `yaml:",inline"`

-	// File is the path to the file targetted by this resource.
+	// File is the path to the file targeted by this resource.
 	File string `yaml:"file"`

 	// Lens is the lens used by this resource. If specified, mgmt
@@ -93,7 +93,7 @@ func (obj *AugeasRes) Validate() error {

 // Init initiates the resource.
 func (obj *AugeasRes) Init() error {
-	obj.BaseRes.kind = "Augeas"
+	obj.BaseRes.kind = "augeas"
 	return obj.BaseRes.Init() // call base init, b/c we're overriding
 }

--- a/resources/exec.go
+++ b/resources/exec.go
@@ -74,7 +74,7 @@ func (obj *ExecRes) Validate() error {

 // Init runs some startup code for this resource.
 func (obj *ExecRes) Init() error {
-	obj.BaseRes.kind = "Exec"
+	obj.BaseRes.kind = "exec"
 	return obj.BaseRes.Init() // call base init, b/c we're overriding
 }

--- a/resources/file.go
+++ b/resources/file.go
@@ -147,7 +147,7 @@ func (obj *FileRes) Init() error {
 	obj.path = obj.GetPath()                     // compute once
 	obj.isDir = strings.HasSuffix(obj.path, "/") // dirs have trailing slashes

-	obj.BaseRes.kind = "File"
+	obj.BaseRes.kind = "file"
 	return obj.BaseRes.Init() // call base init, b/c we're overriding
 }

--- a/resources/hostname.go
+++ b/resources/hostname.go
@@ -87,7 +87,7 @@ func (obj *HostnameRes) Validate() error {

 // Init runs some startup code for this resource.
 func (obj *HostnameRes) Init() error {
-	obj.BaseRes.kind = "Hostname"
+	obj.BaseRes.kind = "hostname"
 	if obj.PrettyHostname == "" {
 		obj.PrettyHostname = obj.Hostname
 	}
--- a/resources/kv.go
+++ b/resources/kv.go
@@ -0,0 +1,304 @@
+// Mgmt
+// Copyright (C) 2013-2017+ 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"
+	"fmt"
+	"log"
+	"strconv"
+
+	errwrap "github.com/pkg/errors"
+)
+
+func init() {
+	gob.Register(&KVRes{})
+}
+
+// KVResSkipCmpStyle represents the different styles of comparison when using SkipLessThan.
+type KVResSkipCmpStyle int
+
+// These are the different allowed comparison styles. Most folks will want SkipCmpStyleInt.
+const (
+	SkipCmpStyleInt KVResSkipCmpStyle = iota
+	SkipCmpStyleString
+)
+
+// KVRes is a resource which writes a key/value pair into cluster wide storage.
+// It will ensure that the key is set to the requested value. The one exception
+// is that if you use the SkipLessThan parameter, then it will only replace the
+// stored value with the requested value if it is greater than that stored one.
+// This allows the KV resource to be used in fast acting, finite state machines
+// which have monotonically increasing state values that represent progression.
+// The one exception is that when this resource receives a refresh signal, then
+// it will set the value to be the exact one if they are not identical already.
+type KVRes struct {
+	BaseRes      `yaml:",inline"`
+	Key          string            `yaml:"key"`          // key to set
+	Value        *string           `yaml:"value"`        // value to set (nil to delete)
+	SkipLessThan bool              `yaml:"skiplessthan"` // skip updates as long as stored value is greater
+	SkipCmpStyle KVResSkipCmpStyle `yaml:"skipcmpstyle"` // how to do the less than cmp
+	// TODO: does it make sense to have different backends here? (eg: local)
+}
+
+// Default returns some sensible defaults for this resource.
+func (obj *KVRes) Default() Res {
+	return &KVRes{
+		BaseRes: BaseRes{
+			MetaParams: DefaultMetaParams, // force a default
+		},
+	}
+}
+
+// Validate if the params passed in are valid data.
+// FIXME: This will catch most issues unless data is passed in after Init with
+// the Send/Recv mechanism. Should the engine re-call Validate after Send/Recv?
+func (obj *KVRes) Validate() error {
+	if obj.Key == "" {
+		return fmt.Errorf("key must not be empty")
+	}
+	if obj.SkipLessThan {
+		if obj.SkipCmpStyle != SkipCmpStyleInt && obj.SkipCmpStyle != SkipCmpStyleString {
+			return fmt.Errorf("the SkipCmpStyle of %v is invalid", obj.SkipCmpStyle)
+		}
+
+		if v := obj.Value; obj.SkipCmpStyle == SkipCmpStyleInt && v != nil {
+			if _, err := strconv.Atoi(*v); err != nil {
+				return fmt.Errorf("the set value of %v can't convert to int", v)
+			}
+		}
+	}
+	return obj.BaseRes.Validate()
+}
+
+// Init initializes the resource.
+func (obj *KVRes) Init() error {
+	obj.BaseRes.kind = "kv"
+	return obj.BaseRes.Init() // call base init, b/c we're overriding
+}
+
+// Watch is the primary listener for this resource and it outputs events.
+func (obj *KVRes) Watch() error {
+
+	// notify engine that we're running
+	if err := obj.Running(); err != nil {
+		return err // bubble up a NACK...
+	}
+
+	ch := obj.Data().World.StrWatch(obj.Key) // get possible events!
+
+	var send = false // send event?
+	var exit *error
+	for {
+		select {
+		// NOTE: this part is very similar to the file resource code
+		case err, ok := <-ch:
+			if !ok { // channel shutdown
+				return nil
+			}
+			if err != nil {
+				return errwrap.Wrapf(err, "unknown %s[%s] watcher error", obj.Kind(), obj.GetName())
+			}
+			if obj.Data().Debug {
+				log.Printf("%s[%s]: Event!", obj.Kind(), obj.GetName())
+			}
+			send = true
+			obj.StateOK(false) // dirty
+
+		case event := <-obj.Events():
+			// we avoid sending events on unpause
+			if exit, send = obj.ReadEvent(event); exit != nil {
+				return *exit // exit
+			}
+		}
+
+		// do all our event sending all together to avoid duplicate msgs
+		if send {
+			send = false
+			obj.Event()
+		}
+	}
+}
+
+// lessThanCheck checks for less than validity.
+func (obj *KVRes) lessThanCheck(value string) (checkOK bool, err error) {
+
+	v := *obj.Value
+	if value == v { // redundant check for safety
+		return true, nil
+	}
+
+	var refresh = obj.Refresh()       // do we have a pending reload to apply?
+	if !obj.SkipLessThan || refresh { // update lessthan on refresh
+		return false, nil
+	}
+
+	switch obj.SkipCmpStyle {
+	case SkipCmpStyleInt:
+		intValue, err := strconv.Atoi(value)
+		if err != nil {
+			// NOTE: We don't error here since we're going to write
+			// over the value anyways. It could be from an old run!
+			return false, nil // value is bad (old/corrupt), fix it
+		}
+		if vint, err := strconv.Atoi(v); err != nil {
+			return false, errwrap.Wrapf(err, "can't convert %v to int", v)
+		} else if vint < intValue {
+			return true, nil
+		}
+
+	case SkipCmpStyleString:
+		if v < value { // weird way to cmp, but valid
+			return true, nil
+		}
+
+	default:
+		return false, fmt.Errorf("unmatches SkipCmpStyle style %v", obj.SkipCmpStyle)
+	}
+
+	return false, nil
+}
+
+// CheckApply method for Password resource. Does nothing, returns happy!
+func (obj *KVRes) CheckApply(apply bool) (checkOK bool, err error) {
+	log.Printf("%s[%s]: CheckApply(%t)", obj.Kind(), obj.GetName(), apply)
+
+	if val, exists := obj.Recv["Value"]; exists && val.Changed {
+		// if we received on Value, and it changed, wooo, nothing to do.
+		log.Printf("CheckApply: `Value` was updated!")
+	}
+
+	hostname := obj.Data().Hostname // me
+	keyMap, err := obj.Data().World.StrGet(obj.Key)
+	if err != nil {
+		return false, errwrap.Wrapf(err, "check error during StrGet")
+	}
+
+	if value, ok := keyMap[hostname]; ok && obj.Value != nil {
+		if value == *obj.Value {
+			return true, nil
+		}
+
+		if c, err := obj.lessThanCheck(value); err != nil {
+			return false, err
+		} else if c {
+			return true, nil
+		}
+
+	} else if !ok && obj.Value == nil {
+		return true, nil // nothing to delete, we're good!
+
+	} else if ok && obj.Value == nil { // delete
+		err := obj.Data().World.StrDel(obj.Key)
+		return false, errwrap.Wrapf(err, "apply error during StrDel")
+	}
+
+	if !apply {
+		return false, nil
+	}
+
+	if err := obj.Data().World.StrSet(obj.Key, *obj.Value); err != nil {
+		return false, errwrap.Wrapf(err, "apply error during StrSet")
+	}
+
+	return false, nil
+}
+
+// KVUID is the UID struct for KVRes.
+type KVUID struct {
+	BaseUID
+	name string
+}
+
+// AutoEdges returns the AutoEdge interface. In this case no autoedges are used.
+func (obj *KVRes) AutoEdges() AutoEdge {
+	return nil
+}
+
+// UIDs includes all params to make a unique identification of this object.
+// Most resources only return one, although some resources can return multiple.
+func (obj *KVRes) UIDs() []ResUID {
+	x := &KVUID{
+		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 *KVRes) GroupCmp(r Res) bool {
+	_, ok := r.(*KVRes)
+	if !ok {
+		return false
+	}
+	return false // TODO: this is doable!
+	// TODO: it could be useful to group our writes and watches!
+}
+
+// Compare two resources and return if they are equivalent.
+func (obj *KVRes) Compare(res Res) bool {
+	switch res.(type) {
+	// we can only compare KVRes to others of the same resource
+	case *KVRes:
+		res := res.(*KVRes)
+		if !obj.BaseRes.Compare(res) { // call base Compare
+			return false
+		}
+
+		if obj.Key != res.Key {
+			return false
+		}
+		if (obj.Value == nil) != (res.Value == nil) { // xor
+			return false
+		}
+		if obj.Value != nil && res.Value != nil {
+			if *obj.Value != *res.Value { // compare the strings
+				return false
+			}
+		}
+		if obj.SkipLessThan != res.SkipLessThan {
+			return false
+		}
+		if obj.SkipCmpStyle != res.SkipCmpStyle {
+			return false
+		}
+	default:
+		return false
+	}
+	return true
+}
+
+// UnmarshalYAML is the custom unmarshal handler for this struct.
+// It is primarily useful for setting the defaults.
+func (obj *KVRes) UnmarshalYAML(unmarshal func(interface{}) error) error {
+	type rawRes KVRes // indirection to avoid infinite recursion
+
+	def := obj.Default()    // get the default
+	res, ok := def.(*KVRes) // put in the right format
+	if !ok {
+		return fmt.Errorf("could not convert to KVRes")
+	}
+	raw := rawRes(*res) // convert; the defaults go here
+
+	if err := unmarshal(&raw); err != nil {
+		return err
+	}
+
+	*obj = KVRes(raw) // restore from indirection with type conversion!
+	return nil
+}
--- a/resources/msg.go
+++ b/resources/msg.go
@@ -75,7 +75,7 @@ func (obj *MsgRes) Validate() error {

 // Init runs some startup code for this resource.
 func (obj *MsgRes) Init() error {
-	obj.BaseRes.kind = "Msg"
+	obj.BaseRes.kind = "msg"
 	return obj.BaseRes.Init() // call base init, b/c we're overrriding
 }

--- a/resources/noop.go
+++ b/resources/noop.go
@@ -49,7 +49,7 @@ func (obj *NoopRes) Validate() error {

 // Init runs some startup code for this resource.
 func (obj *NoopRes) Init() error {
-	obj.BaseRes.kind = "Noop"
+	obj.BaseRes.kind = "noop"
 	return obj.BaseRes.Init() // call base init, b/c we're overriding
 }

--- a/resources/nspawn.go
+++ b/resources/nspawn.go
@@ -92,7 +92,7 @@ func (obj *NspawnRes) Init() error {
 	if err := obj.svc.Init(); err != nil {
 		return err
 	}
-	obj.BaseRes.kind = "Nspawn"
+	obj.BaseRes.kind = "nspawn"
 	return obj.BaseRes.Init()
 }

--- a/resources/password.go
+++ b/resources/password.go
@@ -73,7 +73,7 @@ func (obj *PasswordRes) Validate() error {
 // 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 {
-	obj.BaseRes.kind = "Password" // must be set before using VarDir
+	obj.BaseRes.kind = "password" // must be set before using VarDir

 	dir, err := obj.VarDir("")
 	if err != nil {
--- a/resources/pkg.go
+++ b/resources/pkg.go
@@ -66,7 +66,7 @@ func (obj *PkgRes) Validate() error {

 // Init runs some startup code for this resource.
 func (obj *PkgRes) Init() error {
-	obj.BaseRes.kind = "Pkg"
+	obj.BaseRes.kind = "pkg"
 	if err := obj.BaseRes.Init(); err != nil { // call base init, b/c we're overriding
 		return err
 	}
--- a/resources/resources.go
+++ b/resources/resources.go
@@ -19,14 +19,12 @@
 package resources

 import (
-	"bytes"
-	"encoding/base64"
-	"encoding/gob"
 	"fmt"
 	"log"
 	"math"
 	"os"
 	"path"
+	"sort"
 	"sync"
 	"time"

@@ -34,6 +32,7 @@ import (
 	"github.com/purpleidea/mgmt/converger"
 	"github.com/purpleidea/mgmt/event"
 	"github.com/purpleidea/mgmt/prometheus"
+	"github.com/purpleidea/mgmt/util"

 	errwrap "github.com/pkg/errors"
 	"golang.org/x/time/rate"
@@ -54,12 +53,27 @@ const (

 const refreshPathToken = "refresh"

+// World is an interface to the rest of the different graph state. It allows
+// the GAPI to store state and exchange information throughout the cluster. It
+// is the interface each machine uses to communicate with the rest of the world.
+type World interface { // TODO: is there a better name for this interface?
+	ResExport([]Res) error
+	// FIXME: should this method take a "filter" data struct instead of many args?
+	ResCollect(hostnameFilter, kindFilter []string) ([]Res, error)
+
+	StrWatch(namespace string) chan error
+	StrGet(namespace string) (map[string]string, error)
+	StrSet(namespace, value string) error
+	StrDel(namespace string) error
+}
+
 // Data is the set of input values passed into the pgraph for the resources.
 type Data struct {
-	//Hostname string         // uuid for the host
+	Hostname string // uuid for the host
 	//Noop     bool
 	Converger  converger.Converger
 	Prometheus *prometheus.Prometheus
+	World      World
 	Prefix     string // the prefix to be used for the pgraph namespace
 	Debug      bool
 	// NOTE: we can add more fields here if needed for the resources.
@@ -101,6 +115,7 @@ type MetaParams struct {
 	Poll  uint32     `yaml:"poll"`  // metaparam, number of seconds between poll intervals, 0 to watch
 	Limit rate.Limit `yaml:"limit"` // metaparam, number of events per second to allow through
 	Burst int        `yaml:"burst"` // metaparam, number of events to allow in a burst
+	Sema  []string   `yaml:"sema"`  // metaparam, list of semaphore ids (id | id:count)
 }

 // UnmarshalYAML is the custom unmarshal handler for the MetaParams struct. It
@@ -127,6 +142,7 @@ var DefaultMetaParams = MetaParams{
 	Poll:      0,        // defaults to watching for events
 	Limit:     rate.Inf, // defaults to no limit
 	Burst:     0,        // no burst needed on an infinite rate // TODO: is this a good default?
+	//Sema:      []string{},
 }

 // The Base interface is everything that is common to all resources.
@@ -138,8 +154,10 @@ type Base interface {
 	Kind() string
 	Meta() *MetaParams
 	Events() chan *event.Event
-	AssociateData(*Data)
+	Data() *Data
 	IsWorking() bool
+	IsQuiescing() bool
+	QuiesceGroup() *sync.WaitGroup
 	WaitGroup() *sync.WaitGroup
 	Setup()
 	Reset()
@@ -196,6 +214,7 @@ type BaseRes struct {
 	Recv       map[string]*Send // mapping of key to receive on from value

 	kind   string
+	data   Data
 	state  ResState
 	prefix string // base prefix for this resource

@@ -218,6 +237,8 @@ type BaseRes struct {
 	isStarted bool          // did the started chan already close?
 	starter   bool          // does this have indegree == 0 ? XXX: usually?

+	quiescing    bool // are we quiescing (pause or exit)
+	quiesceGroup *sync.WaitGroup
 	waitGroup    *sync.WaitGroup
 	working      bool // is the Worker() loop running ?
 	debug        bool
@@ -314,19 +335,18 @@ func (obj *BaseRes) Init() error {
 		return fmt.Errorf("resource did not set kind")
 	}

-	obj.cuid = obj.converger.Register()
-	obj.wcuid = obj.converger.Register() // get a cuid for the worker!
-	obj.pcuid = obj.converger.Register() // get a cuid for the process
-
-	obj.eventsLock = &sync.Mutex{}
-	obj.eventsDone = false
-	obj.eventsChan = make(chan *event.Event) // unbuffered chan to avoid stale events
+	obj.cuid = obj.Converger().Register()
+	obj.wcuid = obj.Converger().Register() // get a cuid for the worker!
+	obj.pcuid = obj.Converger().Register() // get a cuid for the process

 	obj.processLock = &sync.Mutex{} // lock around processChan closing and sending
 	obj.processDone = false         // did we close processChan ?
 	obj.processChan = make(chan *event.Event)
 	obj.processSync = &sync.WaitGroup{}

+	obj.quiescing = false // no quiesce operation is happening at the moment
+	obj.quiesceGroup = &sync.WaitGroup{}
+
 	obj.waitGroup = &sync.WaitGroup{} // Init and Close must be 1-1 matched!
 	obj.waitGroup.Add(1)
 	obj.working = true // Worker method should now be running...
@@ -396,12 +416,9 @@ func (obj *BaseRes) Events() chan *event.Event {
 	return obj.eventsChan
 }

-// AssociateData associates some data with the object in question.
-func (obj *BaseRes) AssociateData(data *Data) {
-	obj.converger = data.Converger
-	obj.prometheus = data.Prometheus
-	obj.prefix = data.Prefix
-	obj.debug = data.Debug
+// Data returns an associable handle to some data passed in to the resource.
+func (obj *BaseRes) Data() *Data {
+	return &obj.data
 }

 // IsWorking tells us if the Worker() function is running. Not thread safe.
@@ -409,6 +426,14 @@ func (obj *BaseRes) IsWorking() bool {
 	return obj.working
 }

+// IsQuiescing returns if there is a quiesce operation in progress. Pause and
+// exit both meet this criteria, and this tells some systems to wind down, such
+// as the event replay mechanism.
+func (obj *BaseRes) IsQuiescing() bool { return obj.quiescing }
+
+// QuiesceGroup returns the sync group associated with the quiesce operations.
+func (obj *BaseRes) QuiesceGroup() *sync.WaitGroup { return obj.quiesceGroup }
+
 // WaitGroup returns a sync.WaitGroup which is open when the resource is done.
 // This is more useful than a closed channel signal, since it can be re-used
 // safely without having to recreate it and worry about stale channel handles.
@@ -419,7 +444,10 @@ func (obj *BaseRes) WaitGroup() *sync.WaitGroup { return obj.waitGroup }
 func (obj *BaseRes) Setup() {
 	obj.started = make(chan struct{}) // closes when started
 	obj.stopped = make(chan struct{}) // closes when stopped
-	return
+
+	obj.eventsLock = &sync.Mutex{}
+	obj.eventsDone = false
+	obj.eventsChan = make(chan *event.Event) // unbuffered chan to avoid stale events
 }

 // Reset from Setup.
@@ -430,7 +458,7 @@ func (obj *BaseRes) Reset() {
 // Converger returns the converger object used by the system. It can be used to
 // register new convergers if needed.
 func (obj *BaseRes) Converger() converger.Converger {
-	return obj.converger
+	return obj.data.Converger
 }

 // ConvergerUIDs returns the ConvergerUIDs for the resource. This is called by
@@ -493,6 +521,12 @@ func (obj *BaseRes) GroupRes(res Res) error {
 		return fmt.Errorf("the %v resource is already grouped", res)
 	}

+	// merging two resources into one should yield the sum of their semas
+	if semas := res.Meta().Sema; len(semas) > 0 {
+		obj.Meta().Sema = append(obj.Meta().Sema, semas...)
+		obj.Meta().Sema = util.StrRemoveDuplicatesInList(obj.Meta().Sema)
+	}
+
 	obj.grouped = append(obj.grouped, res)
 	res.SetGrouped(true) // i am contained _in_ a group
 	return nil
@@ -550,6 +584,24 @@ func (obj *BaseRes) Compare(res Res) bool {
 	if obj.Meta().Burst != res.Meta().Burst {
 		return false
 	}
+
+	// are the two slices the same?
+	cmpSlices := func(a, b []string) bool {
+		if len(a) != len(b) {
+			return false
+		}
+		sort.Strings(a)
+		sort.Strings(b)
+		for i := range a {
+			if a[i] != b[i] {
+				return false
+			}
+		}
+		return true
+	}
+	if !cmpSlices(obj.Meta().Sema, res.Meta().Sema) {
+		return false
+	}
 	return true
 }

@@ -633,35 +685,3 @@ func (obj *BaseRes) Poll() error {
 func (obj *BaseRes) Prometheus() *prometheus.Prometheus {
 	return obj.prometheus
 }
-
-// ResToB64 encodes a resource to a base64 encoded string (after serialization)
-func ResToB64(res Res) (string, error) {
-	b := bytes.Buffer{}
-	e := gob.NewEncoder(&b)
-	err := e.Encode(&res) // pass with &
-	if err != nil {
-		return "", fmt.Errorf("Gob failed to encode: %v", err)
-	}
-	return base64.StdEncoding.EncodeToString(b.Bytes()), nil
-}
-
-// B64ToRes decodes a resource from a base64 encoded string (after deserialization)
-func B64ToRes(str string) (Res, error) {
-	var output interface{}
-	bb, err := base64.StdEncoding.DecodeString(str)
-	if err != nil {
-		return nil, fmt.Errorf("Base64 failed to decode: %v", err)
-	}
-	b := bytes.NewBuffer(bb)
-	d := gob.NewDecoder(b)
-	err = d.Decode(&output) // pass with &
-	if err != nil {
-		return nil, fmt.Errorf("Gob failed to decode: %v", err)
-	}
-	res, ok := output.(Res)
-	if !ok {
-		return nil, fmt.Errorf("Output %v is not a Res", res)
-
-	}
-	return res, nil
-}
--- a/resources/sendrecv.go
+++ b/resources/sendrecv.go
@@ -36,6 +36,7 @@ func (obj *BaseRes) Event() error {
 		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()
@@ -69,37 +70,63 @@ func (obj *BaseRes) SendEvent(ev event.Kind, err error) error {
 // 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()
+	//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:
-		// wait for next event to continue
+		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[%s]: unknown event: %v, while paused", obj.Kind(), obj.GetName(), e)
@@ -107,6 +134,7 @@ func (obj *BaseRes) ReadEvent(ev *event.Event) (exit *error, send bool) {
 				//return &err, false
 			}
 		}
+	}
 	err = fmt.Errorf("unknown event: %v", ev)
 	panic(err) // TODO: return a special sentinel instead?
 	//return &err, false
@@ -117,7 +145,7 @@ func (obj *BaseRes) ReadEvent(ev *event.Event) (exit *error, send bool) {
 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 is a very unlikey race that could cause an early converge if the
+	// 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...
--- a/resources/svc.go
+++ b/resources/svc.go
@@ -65,7 +65,7 @@ func (obj *SvcRes) Validate() error {

 // Init runs some startup code for this resource.
 func (obj *SvcRes) Init() error {
-	obj.BaseRes.kind = "Svc"
+	obj.BaseRes.kind = "svc"
 	return obj.BaseRes.Init() // call base init, b/c we're overriding
 }

@@ -174,6 +174,8 @@ func (obj *SvcRes) Watch() error {
 						log.Printf("Svc[%s]->Stopped", svc)
 					case "reloading":
 						log.Printf("Svc[%s]->Reloading", svc)
+					case "failed":
+						log.Printf("Svc[%s]->Failed", svc)
 					default:
 						log.Fatalf("Unknown svc state: %s", event[svc].ActiveState)
 					}
--- a/resources/timer.go
+++ b/resources/timer.go
@@ -58,7 +58,7 @@ func (obj *TimerRes) Validate() error {

 // Init runs some startup code for this resource.
 func (obj *TimerRes) Init() error {
-	obj.BaseRes.kind = "Timer"
+	obj.BaseRes.kind = "timer"
 	return obj.BaseRes.Init() // call base init, b/c we're overrriding
 }

--- a/resources/util.go
+++ b/resources/util.go
@@ -0,0 +1,59 @@
+// Mgmt
+// Copyright (C) 2013-2017+ 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 (
+	"bytes"
+	"encoding/base64"
+	"encoding/gob"
+	"fmt"
+
+	errwrap "github.com/pkg/errors"
+)
+
+// ResToB64 encodes a resource to a base64 encoded string (after serialization).
+func ResToB64(res Res) (string, error) {
+	b := bytes.Buffer{}
+	e := gob.NewEncoder(&b)
+	err := e.Encode(&res) // pass with &
+	if err != nil {
+		return "", errwrap.Wrapf(err, "gob failed to encode")
+	}
+	return base64.StdEncoding.EncodeToString(b.Bytes()), nil
+}
+
+// B64ToRes decodes a resource from a base64 encoded string (after deserialization).
+func B64ToRes(str string) (Res, error) {
+	var output interface{}
+	bb, err := base64.StdEncoding.DecodeString(str)
+	if err != nil {
+		return nil, errwrap.Wrapf(err, "base64 failed to decode")
+	}
+	b := bytes.NewBuffer(bb)
+	d := gob.NewDecoder(b)
+	err = d.Decode(&output) // pass with &
+	if err != nil {
+		return nil, errwrap.Wrapf(err, "gob failed to decode")
+	}
+	res, ok := output.(Res)
+	if !ok {
+		return nil, fmt.Errorf("Output %v is not a Res", res)
+
+	}
+	return res, nil
+}
--- a/resources/virt.go
+++ b/resources/virt.go
@@ -191,7 +191,7 @@ func (obj *VirtRes) Init() error {
 		}
 	}
 	obj.wg = &sync.WaitGroup{}
-	obj.BaseRes.kind = "Virt"
+	obj.BaseRes.kind = "virt"
 	return obj.BaseRes.Init() // call base init, b/c we're overriding
 }

--- a/tag.sh
+++ b/tag.sh
@@ -7,5 +7,5 @@ echo "Version $t is now tagged!"
 echo "Pushing $t to origin..."
 echo "Press ^C within 3s to abort."
 sleep 3s
-git tag $t
+echo "release: tag $t" | git tag --file=- --sign $t
 git push origin $t
--- a/test/shell/file-move.sh
+++ b/test/shell/file-move.sh
@@ -8,19 +8,19 @@ timeout --kill-after=20s 15s ./mgmt run --tmp-prefix --yaml=file-move.yaml 2>&1
 pid=$!
 sleep 5s	# let it converge

-initial=$(grep -c 'File\[file1\]: contentCheckApply(true)' /tmp/mgmt/file-move.log)
+initial=$(grep -c 'file\[file1\]: contentCheckApply(true)' /tmp/mgmt/file-move.log)

 mv /tmp/mgmt/f1 /tmp/mgmt/f2

 sleep 3

-after_move_count=$(grep -c 'File\[file1\]: contentCheckApply(true)' /tmp/mgmt/file-move.log)
+after_move_count=$(grep -c 'file\[file1\]: contentCheckApply(true)' /tmp/mgmt/file-move.log)

 sleep 3

 echo f2 > /tmp/mgmt/f2

-after_moved_file_count=$(grep -c 'File\[file1\]: contentCheckApply(true)' /tmp/mgmt/file-move.log)
+after_moved_file_count=$(grep -c 'file\[file1\]: contentCheckApply(true)' /tmp/mgmt/file-move.log)


 if [[ ${after_move_count} -le ${initial} ]]
--- a/test/shell/graph-fanin-1.sh
+++ b/test/shell/graph-fanin-1.sh
@@ -0,0 +1,7 @@
+#!/bin/bash -e
+
+# should take slightly more than 25s, but fail if we take 35s)
+$timeout --kill-after=35s 30s ./mgmt run --yaml graph-fanin-1.yaml --converged-timeout=5 --no-watch --tmp-prefix --no-pgp &
+pid=$!
+wait $pid	# get exit status
+exit $?
--- a/test/shell/graph-fanin-1.yaml
+++ b/test/shell/graph-fanin-1.yaml
@@ -1,6 +1,6 @@
 ---
 graph: mygraph
-comment: simple exec fan in example to demonstrate optimization)
+comment: simple exec fan in example to demonstrate optimization
 resources:
  exec:
  - name: exec1
--- a/test/shell/prometheus-3.sh
+++ b/test/shell/prometheus-3.sh
@@ -1,5 +1,7 @@
 #!/bin/bash -e

+exit 0	# FIXME: disabled until intermittent failures can be resolved
+
 # run a graph, with prometheus support
 timeout --kill-after=30s 25s ./mgmt run --tmp-prefix --no-pgp --prometheus --yaml prometheus-3.yaml &
 pid=$!
@@ -9,10 +11,13 @@ sleep 10s	# let it converge
 curl 127.0.0.1:9233/metrics

 # Three CheckApply for a File ; with events
-curl 127.0.0.1:9233/metrics | grep '^mgmt_checkapply_total{apply="true",errorful="false",eventful="true",kind="File"} 3$'
+curl 127.0.0.1:9233/metrics | grep '^mgmt_checkapply_total{apply="true",errorful="false",eventful="true",kind="file"} 3$'

 # One CheckApply for a File ; in noop mode.
-curl 127.0.0.1:9233/metrics | grep '^mgmt_checkapply_total{apply="false",errorful="false",eventful="true",kind="File"} 1$'
+curl 127.0.0.1:9233/metrics | grep '^mgmt_checkapply_total{apply="false",errorful="false",eventful="true",kind="file"} 1$'
+
+# Check mgmt_graph_start_time_seconds
+curl 127.0.0.1:9233/metrics | grep "^mgmt_graph_start_time_seconds [1-9]\+"

 killall -SIGINT mgmt	# send ^C to exit mgmt
 wait $pid	# get exit status
--- a/test/shell/sema-1.sh
+++ b/test/shell/sema-1.sh
@@ -0,0 +1,8 @@
+#!/bin/bash -e
+
+# should take at least 55s, but fail if we block this
+# TODO: it would be nice to make sure this test doesn't exit too early!
+$timeout --kill-after=120s 110s ./mgmt run --yaml sema-1.yaml --sema 2 --converged-timeout=5 --no-watch --no-pgp --tmp-prefix &
+pid=$!
+wait $pid	# get exit status
+exit $?
--- a/test/shell/sema-1.yaml
+++ b/test/shell/sema-1.yaml
@@ -0,0 +1,128 @@
+---
+graph: mygraph
+comment: simple exec fan in to fan out example with semaphore
+resources:
+  exec:
+  - name: exec1
+    cmd: sleep 10s
+    shell: ''
+    timeout: 0
+    watchcmd: ''
+    watchshell: ''
+    ifcmd: ''
+    ifshell: ''
+    pollint: 0
+    state: present
+  - name: exec2
+    cmd: sleep 10s
+    shell: ''
+    timeout: 0
+    watchcmd: ''
+    watchshell: ''
+    ifcmd: ''
+    ifshell: ''
+    pollint: 0
+    state: present
+  - name: exec3
+    cmd: sleep 10s
+    shell: ''
+    timeout: 0
+    watchcmd: ''
+    watchshell: ''
+    ifcmd: ''
+    ifshell: ''
+    pollint: 0
+    state: present
+  - name: exec4
+    cmd: sleep 10s
+    shell: ''
+    timeout: 0
+    watchcmd: ''
+    watchshell: ''
+    ifcmd: ''
+    ifshell: ''
+    pollint: 0
+    state: present
+  - name: exec5
+    cmd: sleep 10s
+    shell: ''
+    timeout: 0
+    watchcmd: ''
+    watchshell: ''
+    ifcmd: ''
+    ifshell: ''
+    pollint: 0
+    state: present
+  - name: exec6
+    cmd: sleep 10s
+    shell: ''
+    timeout: 0
+    watchcmd: ''
+    watchshell: ''
+    ifcmd: ''
+    ifshell: ''
+    pollint: 0
+    state: present
+  - name: exec7
+    cmd: sleep 10s
+    shell: ''
+    timeout: 0
+    watchcmd: ''
+    watchshell: ''
+    ifcmd: ''
+    ifshell: ''
+    pollint: 0
+    state: present
+  - name: exec8
+    cmd: sleep 10s
+    shell: ''
+    timeout: 0
+    watchcmd: ''
+    watchshell: ''
+    ifcmd: ''
+    ifshell: ''
+    pollint: 0
+    state: present
+edges:
+- name: e1
+  from:
+    kind: exec
+    name: exec1
+  to:
+    kind: exec
+    name: exec4
+- name: e2
+  from:
+    kind: exec
+    name: exec2
+  to:
+    kind: exec
+    name: exec4
+- name: e3
+  from:
+    kind: exec
+    name: exec3
+  to:
+    kind: exec
+    name: exec4
+- name: e4
+  from:
+    kind: exec
+    name: exec4
+  to:
+    kind: exec
+    name: exec5
+- name: e5
+  from:
+    kind: exec
+    name: exec4
+  to:
+    kind: exec
+    name: exec6
+- name: e6
+  from:
+    kind: exec
+    name: exec4
+  to:
+    kind: exec
+    name: exec7
--- a/test/shell/t4.sh
+++ b/test/shell/t4.sh
@@ -1,7 +0,0 @@
-#!/bin/bash -e
-
-# should take slightly more than 25s, but fail if we take 35s)
-$timeout --kill-after=35s 30s ./mgmt run --yaml t4.yaml --converged-timeout=5 --no-watch --tmp-prefix &
-pid=$!
-wait $pid	# get exit status
-exit $?
--- a/test/test-golint.sh
+++ b/test/test-golint.sh
@@ -28,7 +28,7 @@ COUNT=`echo -e "$LINT" | wc -l`	# number of golint problems in current branch
 [ "$LINT" = "" ] && echo PASS && exit	# everything is "perfect"
 echo "$LINT"	# display the issues

-T=`mktemp --tmpdir -d tmp.XXX`
+T=`mktemp --tmpdir -d tmp.X'X'X`	# add quotes to avoid matching three X's
 [ "$T" = "" ] && fail_test "Could not create tmpdir"
 cd $T || fail_test "Could not change into tmpdir $T"
 git clone --recursive "${ROOT}" 2>/dev/null	# make a copy
--- a/test/test-govet.sh
+++ b/test/test-govet.sh
@@ -22,7 +22,8 @@ function simplify-gocase() {
 }

 function token-coloncheck() {
-	if grep -Ei "[\/]+[\/]+[ ]*+(FIXME[^:]|TODO[^:]|XXX[^:])" "$1"; then
+	# add quotes to avoid matching three X's
+	if grep -Ei "[\/]+[\/]+[ ]*+(FIXME[^:]|TODO[^:]|X"'X'"X[^:])" "$1"; then
 		return 1	# tokens must end with a colon
 	fi
 	return 0
--- a/test/test-reproducible.sh
+++ b/test/test-reproducible.sh
@@ -7,7 +7,7 @@ set -o pipefail
 DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && cd .. && pwd )"	# dir!
 cd "$DIR" >/dev/null	# work from main mgmt directory
 make build
-T=`mktemp --tmpdir -d tmp.XXX`
+T=`mktemp --tmpdir -d tmp.X'X'X`	# add quotes to avoid matching three X's
 cp -a ./mgmt "$T"/mgmt.1
 make clean
 make build
--- a/util/semaphore/semaphore.go
+++ b/util/semaphore/semaphore.go
@@ -0,0 +1,77 @@
+// Mgmt
+// Copyright (C) 2013-2017+ 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 semaphore contains an implementation of a counting semaphore.
+package semaphore
+
+import (
+	"fmt"
+)
+
+// Semaphore is a counting semaphore. It must be initialized before use.
+type Semaphore struct {
+	C      chan struct{}
+	closed chan struct{}
+}
+
+// NewSemaphore creates a new semaphore.
+func NewSemaphore(size int) *Semaphore {
+	obj := &Semaphore{}
+	obj.Init(size)
+	return obj
+}
+
+// Init initializes the semaphore.
+func (obj *Semaphore) Init(size int) {
+	obj.C = make(chan struct{}, size)
+	obj.closed = make(chan struct{})
+}
+
+// Close shuts down the semaphore and releases all the locks.
+func (obj *Semaphore) Close() {
+	// TODO: we could return an error if any semaphores were killed, but
+	// it's not particularly useful to know that for this application...
+	close(obj.closed)
+}
+
+// P acquires n resources.
+func (obj *Semaphore) P(n int) error {
+	for i := 0; i < n; i++ {
+		select {
+		case obj.C <- struct{}{}: // acquire one
+		case <-obj.closed: // exit signal
+			return fmt.Errorf("closed")
+		}
+	}
+	return nil
+}
+
+// V releases n resources.
+func (obj *Semaphore) V(n int) error {
+	for i := 0; i < n; i++ {
+		select {
+		case <-obj.C: // release one
+		// TODO: is the closed signal needed if unlocks should always pass?
+		case <-obj.closed: // exit signal
+			return fmt.Errorf("closed")
+		// TODO: is it true you shouldn't call a release before a lock?
+		default: // trying to release something that isn't locked
+			panic("semaphore: V > P")
+		}
+	}
+	return nil
+}
--- a/util/util.go
+++ b/util/util.go
@@ -57,7 +57,7 @@ func Uint64KeyFromStrInMap(needle string, haystack map[uint64]string) (uint64, b
 }

 // StrRemoveDuplicatesInList removes any duplicate values in the list.
-// This is a possibly sub-optimal, O(n^2)? implementation.
+// This implementation is possibly sub-optimal (O(n^2)?) but preserves ordering.
 func StrRemoveDuplicatesInList(list []string) []string {
 	unique := []string{}
 	for _, x := range list {
--- a/yamlgraph/gconfig.go
+++ b/yamlgraph/gconfig.go
@@ -26,10 +26,8 @@ import (
 	"reflect"
 	"strings"

-	"github.com/purpleidea/mgmt/gapi"
 	"github.com/purpleidea/mgmt/pgraph"
 	"github.com/purpleidea/mgmt/resources"
-	"github.com/purpleidea/mgmt/util"

 	"gopkg.in/yaml.v2"
 )
@@ -60,6 +58,7 @@ type Resources struct {
 	Exec     []*resources.ExecRes     `yaml:"exec"`
 	File     []*resources.FileRes     `yaml:"file"`
 	Hostname []*resources.HostnameRes `yaml:"hostname"`
+	KV       []*resources.KVRes       `yaml:"kv"`
 	Msg      []*resources.MsgRes      `yaml:"msg"`
 	Noop     []*resources.NoopRes     `yaml:"noop"`
 	Nspawn   []*resources.NspawnRes   `yaml:"nspawn"`
@@ -93,7 +92,7 @@ func (c *GraphConfig) Parse(data []byte) error {

 // NewGraphFromConfig transforms a GraphConfig struct into a new graph.
 // FIXME: remove any possibly left over, now obsolete graph diff code from here!
-func (c *GraphConfig) NewGraphFromConfig(hostname string, world gapi.World, noop bool) (*pgraph.Graph, error) {
+func (c *GraphConfig) NewGraphFromConfig(hostname string, world resources.World, noop bool) (*pgraph.Graph, error) {
 	// hostname is the uuid for the host

 	var graph *pgraph.Graph          // new graph to return
@@ -116,8 +115,7 @@ func (c *GraphConfig) NewGraphFromConfig(hostname string, world gapi.World, noop
 		field := value.FieldByName(name)
 		iface := field.Interface() // interface type of value
 		slice := reflect.ValueOf(iface)
-		// XXX: should we just drop these everywhere and have the kind strings be all lowercase?
-		kind := util.FirstToUpper(name)
+		kind := strings.ToLower(name)
 		for j := 0; j < slice.Len(); j++ { // loop through resources of same kind
 			x := slice.Index(j).Interface()
 			res, ok := x.(resources.Res) // convert to Res type
@@ -158,8 +156,7 @@ func (c *GraphConfig) NewGraphFromConfig(hostname string, world gapi.World, noop
 	var hostnameFilter []string // empty to get from everyone
 	kindFilter := []string{}
 	for _, t := range c.Collector {
-		// XXX: should we just drop these everywhere and have the kind strings be all lowercase?
-		kind := util.FirstToUpper(t.Kind)
+		kind := strings.ToLower(t.Kind)
 		kindFilter = append(kindFilter, kind)
 	}
 	// do all the graph look ups in one single step, so that if the backend
@@ -175,8 +172,7 @@ func (c *GraphConfig) NewGraphFromConfig(hostname string, world gapi.World, noop
 		matched := false
 		// see if we find a collect pattern that matches
 		for _, t := range c.Collector {
-			// XXX: should we just drop these everywhere and have the kind strings be all lowercase?
-			kind := util.FirstToUpper(t.Kind)
+			kind := strings.ToLower(t.Kind)
 			// use t.Kind and optionally t.Pattern to collect from storage
 			log.Printf("Collect: %v; Pattern: %v", kind, t.Pattern)

@@ -219,20 +215,20 @@ func (c *GraphConfig) NewGraphFromConfig(hostname string, world gapi.World, noop
 	}

 	for _, e := range c.Edges {
-		if _, ok := lookup[util.FirstToUpper(e.From.Kind)]; !ok {
+		if _, ok := lookup[strings.ToLower(e.From.Kind)]; !ok {
 			return nil, fmt.Errorf("can't find 'from' resource")
 		}
-		if _, ok := lookup[util.FirstToUpper(e.To.Kind)]; !ok {
+		if _, ok := lookup[strings.ToLower(e.To.Kind)]; !ok {
 			return nil, fmt.Errorf("can't find 'to' resource")
 		}
-		if _, ok := lookup[util.FirstToUpper(e.From.Kind)][e.From.Name]; !ok {
+		if _, ok := lookup[strings.ToLower(e.From.Kind)][e.From.Name]; !ok {
 			return nil, fmt.Errorf("can't find 'from' name")
 		}
-		if _, ok := lookup[util.FirstToUpper(e.To.Kind)][e.To.Name]; !ok {
+		if _, ok := lookup[strings.ToLower(e.To.Kind)][e.To.Name]; !ok {
 			return nil, fmt.Errorf("can't find 'to' name")
 		}
-		from := lookup[util.FirstToUpper(e.From.Kind)][e.From.Name]
-		to := lookup[util.FirstToUpper(e.To.Kind)][e.To.Name]
+		from := lookup[strings.ToLower(e.From.Kind)][e.From.Name]
+		to := lookup[strings.ToLower(e.To.Kind)][e.To.Name]
 		edge := pgraph.NewEdge(e.Name)
 		edge.Notify = e.Notify
 		graph.AddEdge(from, to, edge)
Author	SHA1	Message	Date
James Shubin	9546949945	git: Improve tagging script	2017-03-11 08:29:53 -05:00
James Shubin	8ff048d055	test: Disable prometheus-3.sh test temporarily It seems to be failing, and I'm not sure where the regression is, or if there is a race. Sorry roidelapluie.	2017-03-09 11:46:21 -05:00
James Shubin	95a1c6e7fb	pgraph, resources: Discard BackPokes during pause and resume This prevents some nasty races where a BackPoke could arrive on a paused vertex either during a resume or pause operation. Previously we might also have poked an excessive number of resources on resume. The solution was to discard BackPokes during pause or resume. On pause, they can be discarded because we've asked the graph to quiesce, and any further work can be done on resume, and on resume we ignore them because this should only happen during the unrolling (reverse topological resume of the graph) and at the end of this the indegree == 0 vertices will initiate a series of pokes which should deal with any BackPoke that was possibly discarded. One other aspect of this which is important: if an indegree == 0 vertex is poked (Process runs) but it's already in the correct state, it should still transmit the Poke through itself so that subsequent vertices know to run. Currently this is done correctly in Process(). I'm a bit ashamed that this wasn't done properly in the engine earlier, but I suppose that's what comes out of running fancier graphs and really thinking in detail about what's truly correct. Hopefully I got it right this time!	2017-03-09 06:35:15 -05:00
James Shubin	0b1a4a0f30	pgraph, resources: Quiesce when pausing or exiting the resource This prevents a nasty race that can happen in a graph with more than one resource. If a resource has someone that it can BackPoke, and then suppose an event comes in. It runs the obj.Event() method (from inside its Watch loop) and then before the resulting Process method can run it receives a pause event and pauses. Then the parent resource pauses as well. Finally (it's a race) the Process gets around to running, and decides it needs to BackPoke. At this point since the parent resource is paused, it receives the BackPoke at a time when it can't handle receiving one, and it panics! As a result, we now track the number of running Process possibilities via a WaitGroup which gets incremented from the obj.Event() and we don't finish our pause or exit operations until it has quiesced and our WaitGroup lets us know via Wait(). Lastly in order to prevent repeated replays, we detect when we're quiescing and suspend replaying until post pause. We don't need to save the replay (playback variable) explicitly because its state remains during pause, and on exit it would get re-checked anyways.	2017-03-09 02:50:55 -05:00
James Shubin	22b48e296a	resources, yamlgraph: Drop the kind capitalization This stopped making sense now that we have a resource with two primary capitals. It was just a silly formatting hack anyways. Welcome kv!	2017-03-09 02:50:55 -05:00
James Shubin	c696ebf53c	resources: svc: Add failed state Services can be in a failed state too.	2017-03-08 19:23:33 -05:00
James Shubin	a0686b7d2b	pgraph: graphviz: Update Graphviz lib to quote names properly This also moves the library to after the graph starts so that the kind fields will be visible.	2017-03-08 19:23:33 -05:00
James Shubin	8d94be8924	resources: kv: Add new KV resource which sets key value pairs This is a new resource for setting key value pairs in our global world database. Currently only etcd is supported. Some of the implications and possibilities of this resource will become more obvious with future commits! You can bother/test this resource with these commands: ETCDCTL_API=3 etcdctl get "/_mgmt/strings/" --prefix=true ETCDCTL_API=3 etcdctl put "/_mgmt/strings/KEY/HOSTNAME" 42 Replace the KEY and HOSTNAME variables with the actual values you'd like to use. The 42 is the value that is set.	2017-03-08 19:23:33 -05:00
James Shubin	e97ac5033f	resources: Split util functions into separate file This also adds errwrap to their implementation.	2017-03-08 19:23:33 -05:00
James Shubin	44771a0049	gapi: Move the World interface into resources This was necessary to fix some "import cycle" errors I was having when adding the World api to the resource Data struct. I think this is a good hint that I need to start splitting up existing packages into sub files, and cleaning up and inter-package problems too.	2017-03-08 19:23:33 -05:00
James Shubin	32aae8f57a	lib, pgraph, resources: Refactor data association API This should make things cleaner and help avoid as much churn every time we change a property.	2017-03-07 22:51:11 -05:00
James Shubin	8207e23cd9	lib: Refactor instantiation of world API	2017-03-07 22:51:11 -05:00
James Shubin	a469029698	etcd: Switch to buffered channel to remove duplicates Since we don't return the actual values and instead only tell about events (which leaves the `Get` of the value as a second operation) then we don't have to use a channel with backpressure since all the events are identical.	2017-03-07 22:51:11 -05:00
James Shubin	203d866643	gapi, etcd: Define and implement a string sharing API for the World This adds a new set of methods to the World API (for sharing data throughout the cluster) and adds an etcd backed implementation.	2017-03-07 22:51:11 -05:00
Michael Borden	1488e5ec4d	travis: Add go_import_path This should fix an issue with turning on travis for any mgmt fork.	2017-03-07 14:17:53 -08:00
Michael Borden	af66138a17	misc: Add pacman support	2017-03-03 21:04:29 -08:00
James Shubin	5f060d60a7	test: Avoid matching three X's This helps my "WIP" detector script avoid false positives. It is a simple script which helps me find release critical problems.	2017-03-01 22:37:08 -05:00
James Shubin	73ccbb69ea	travis: Ensure recent golang versions in test This ensures travis picks the latest versions. Apparently writing 1.x is also a valid choice.	2017-03-01 21:45:02 -05:00
James Shubin	be60440b20	readme: Add new blog post about metaparameters This also documents the recent semaphore work.	2017-03-01 16:18:14 -05:00
James Shubin	837efb78e6	spelling: Fix typos as found by goreportcard	2017-02-28 23:48:34 -05:00
James Shubin	4a62a290d8	pgraph: Clean up tests This splits the tests into multiple files.	2017-02-28 16:47:16 -05:00
James Shubin	018399cb1f	semaphore, pgraph: Add semaphore grouping and tests If two resources are grouped, then the result should contain the semaphores of both resources. This is because the user is expecting (independently) resource A and resource B to have a limiting choke point. If when combined those choke points aren't preserved, then we have broken an important promise to the user.	2017-02-28 16:40:53 -05:00
James Shubin	646a576358	remote: Replace builtin semaphore type with common util lib Refactor code to use the new fancy lib!	2017-02-27 02:57:06 -05:00
James Shubin	d8e19cd79a	semaphore: Create a semaphore metaparam This adds a P/V style semaphore mechanism to the resource graph. This enables the user to specify a number of "id:count" tags associated with each resource which will reduce the parallelism of the CheckApply operation to that maximum count. This is particularly interesting because (assuming I'm not mistaken) the implementation is dead-lock free assuming that no individual resource permanently ever blocks during execution! I don't have a formal proof of this, but I was able to convince myself on paper that it was the case. An actual proof that N P/V counting semaphores in a DAG won't ever dead-lock would be particularly welcome! Hint: the trick is to acquire them in alphabetical order while respecting the DAG flow. Disclaimer, this assumes that the lock count is always > 0 of course.	2017-02-27 02:57:06 -05:00
James Shubin	757cb0cf23	test: Small fixups to t4 and a rename	2017-02-26 20:54:22 -05:00
Julien Pivotto	7d92ab335a	prometheus: Add mgmt_pgraph_start_time_seconds metric Signed-off-by: Julien Pivotto <roidelapluie@inuits.eu>	2017-02-26 15:28:43 +01:00
James Shubin	46c6d6f656	compiling: Add -i flag to go build to speed up builds So builds take about 30s on my shitty machine which is pretty long. Turns out golang caches things in $GOPATH/pkg/ but is not clever enough to erase things from there that are out of date. As a result, it was rebuilding everything (including the unchanged dependencies) every build! By completely wiping out $GOPATH/pkg/ and then running `go build -i`, this now takes builds down to about 8 seconds. (After one full build is finished.) This is basically the same as running `go install`, but without copying junk to $GOPATH/bin. Hopefully the tooling will be smart enough to know when to throw out stuff in $GOPATH/pkg automatically and avoid this problem entirely. Is it wrong to send Google a bill for all the extra cpu cycles I've used? ;)	2017-02-26 04:06:36 -05:00
Julien Pivotto	46260749c1	prometheus: Move the prometheus nil check inside the prometheus function That pattern will be reused in future metrics. Signed-off-by: Julien Pivotto <roidelapluie@inuits.eu>	2017-02-26 09:33:34 +01:00
Julien Pivotto	50664fe115	compilation: Make build the default target It is really strange that whe I run make, it does not build mgmt. This commit makes build the default target, without moving the target, therefore we keep as much as we can the order of the file. This also removes the confusion for designers that would run "make" instead of "make art", whose work would be disrupted when we add a -- let's say -- make alpharelese command. Signed-off-by: Julien Pivotto <roidelapluie@inuits.eu>	2017-02-26 09:04:20 +01:00