First pass: simplest thing that could possibly work.

Author: relistan

LICENSE

@@ -0,0 +1,19 @@
+Copyright (c) 2015 New Relic, Inc.
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

bosun.go

@@ -0,0 +1,199 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	"github.com/hashicorp/memberlist"
+)
+
+var broadcasts chan [][]byte
+var servicesState map[string]*Server
+
+type Server struct {
+	Name string
+	Services map[string]*ServiceContainer
+	LastUpdated time.Time
+}
+
+func (p *Server) Init(name string) {
+	p.Name = ""
+	// Pre-create for 5 services per host
+	p.Services = make(map[string]*ServiceContainer, 5)
+	p.LastUpdated = time.Unix(0, 0)
+}
+
+type servicesDelegate struct {}
+
+func (d *servicesDelegate) NodeMeta(limit int) []byte {
+	fmt.Printf("NodeMeta(): %d\n", limit)
+	return []byte(`{ "State": "Running" }`)
+}
+
+func (d *servicesDelegate) NotifyMsg(message []byte) {
+	if len(message) <  1 {
+		fmt.Println("NotifyMsg(): empty")
+		return
+	}
+
+	fmt.Printf("NotifyMsg(): %s\n", string(message))
+
+	// TODO don't just send container structs, send message structs
+	data := Decode(message)
+	if data == nil {
+		fmt.Printf("NotifyMsg(): error decoding!\n")
+		return
+	}
+
+	addServiceEntry(*data)
+}
+
+func (d *servicesDelegate) GetBroadcasts(overhead, limit int) [][]byte {
+	fmt.Printf("GetBroadcasts(): %d %d\n", overhead, limit)
+
+	select {
+		case broadcast := <-broadcasts:
+			println("Sending broadcast")
+			return broadcast
+		default:
+			return nil
+	}
+}
+
+func (d *servicesDelegate) LocalState(join bool) []byte {
+	fmt.Printf("LocalState(): %b\n", join)
+	return []byte("some state")
+}
+
+func (d *servicesDelegate) MergeRemoteState(buf []byte, join bool) {
+	fmt.Printf("MergeRemoteState(): %s %b\n", string(buf), join)
+}
+
+func updateState() {
+	for ;; {
+		containerList := containers()
+		prepared := make([][]byte, len(containerList))
+
+		for _, container := range containerList {
+			addServiceEntry(container)
+			encoded, err := container.Encode()
+			if err != nil {
+				log.Printf("ERROR encoding container: (%s)", err.Error())
+				continue
+			}
+
+			prepared = append(prepared, encoded)
+		}
+		broadcasts <- prepared
+
+		time.Sleep(2 * time.Second)
+	}
+}
+
+func updateMetaData(list *memberlist.Memberlist, metaUpdates chan []byte) {
+	for ;; {
+		list.LocalNode().Meta = <-metaUpdates // Blocking
+		fmt.Printf("Got update: %s\n", string(list.LocalNode().Meta))
+		err := list.UpdateNode(10 * time.Second)
+		if err != nil {
+			fmt.Printf("Error pushing node update!")
+		}
+	}
+}
+
+func announceMembers(list *memberlist.Memberlist) {
+	for ;; {
+		// Ask for members of the cluster
+		for _, member := range list.Members() {
+		    fmt.Printf("Member: %s %s\n", member.Name, member.Addr)
+			fmt.Printf("  Meta:\n    %s\n", string(member.Meta))
+		}
+
+		printServices(list);
+
+		time.Sleep(2 * time.Second)
+	}
+}
+
+func formatServices(list *memberlist.Memberlist) string {
+	var output string
+
+	output += "Services ------------------------------\n"
+	for hostname, server := range servicesState {
+		output += fmt.Sprintf("  %s: (%s)\n", hostname, server.LastUpdated.String())
+		for _, service := range server.Services {
+			output += fmt.Sprintf("      %s %-20s %-30s %s\n",
+				service.ID,
+				service.Name,
+				service.Image,
+				service.Created,
+			)
+		}
+		output += "\n"
+	}
+
+	output += "\nCluster Hosts -------------------------\n"
+	for _, host := range list.Members() {
+		output += fmt.Sprintf("    %s\n", host.Name)
+	}
+
+	output += "---------------------------------------"
+
+	return output
+}
+
+func printServices(list *memberlist.Memberlist) {
+	println(formatServices(list))
+}
+
+func addServiceEntry(data ServiceContainer) {
+	// Lazily create the maps
+	if servicesState == nil {
+		// Pre-create for 5 hosts
+		servicesState = make(map[string]*Server, 5)
+	}
+	if servicesState[data.Hostname] == nil {
+		var server Server
+		server.Init(data.Hostname)
+		servicesState[data.Hostname] = &server
+	}
+
+	servicesState[data.Hostname].Services[data.ID] = &data
+	// TODO Use the message time!
+	servicesState[data.Hostname].LastUpdated = time.Now().UTC()
+}
+
+func main() {
+	opts := parseCommandLine()
+
+	var delegate servicesDelegate
+
+	broadcasts = make(chan [][]byte)
+
+	config := memberlist.DefaultLANConfig()
+	config.Delegate = &delegate
+
+	list, err := memberlist.Create(config)
+	exitWithError(err, "Failed to create memberlist")
+
+	// Join an existing cluster by specifying at least one known member.
+	_, err = list.Join([]string{ opts.ClusterIP })
+	exitWithError(err, "Failed to join cluster")
+
+	metaUpdates := make(chan []byte)
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	go announceMembers(list)
+	go updateState()
+	go updateMetaData(list, metaUpdates)
+
+	serveHttp(list)
+
+	time.Sleep(4 * time.Second)
+	metaUpdates <-[]byte("A message!")
+
+	wg.Wait() // forever... nothing will decrement the wg
+}

cli.go

@@ -0,0 +1,30 @@
+package main
+
+import (
+	"errors"
+	"flag"
+	"log"
+)
+
+type CliOpts struct {
+	ClusterIP	 string
+}
+
+func exitWithError(err error, message string) {
+    if err != nil {
+        log.Fatalf("%s (%s)", message, err.Error())
+    }
+}
+
+func parseCommandLine() *CliOpts {
+	var opts CliOpts
+
+	flag.StringVar(&opts.ClusterIP, "cluster-ip", "", "The initial cluster bootstrap IP")
+
+	flag.Parse()
+	if opts.ClusterIP == "" {
+		exitWithError(errors.New(""), "-cluster-ip is required!")
+	}
+
+	return &opts
+}

docker.go

@@ -0,0 +1,60 @@
+package main
+
+import (
+	"encoding/json"
+	"os"
+	"time"
+
+	"github.com/fsouza/go-dockerclient"
+)
+
+type ServiceContainer struct {
+	ID string
+	Name string
+	Image string
+	Created time.Time
+	Hostname string
+}
+
+func (container ServiceContainer) Encode() ([]byte, error) {
+	return json.Marshal(container)
+}
+
+func Decode(data []byte) *ServiceContainer {
+	var container ServiceContainer
+	json.Unmarshal(data, &container)
+
+	return &container
+}
+
+// Format an APIContainers struct into a more compact struct we
+// can ship over the wire in a broadcast.
+func toServiceContainer(container docker.APIContainers) ServiceContainer {
+	var svcContainer ServiceContainer
+	hostname, _ := os.Hostname()
+
+	svcContainer.ID       = container.ID[0:7]  // Use short IDs
+	svcContainer.Name     = container.Names[0] // Use the first name
+	svcContainer.Image    = container.Image
+	svcContainer.Created  = time.Unix(container.Created, 0)
+	svcContainer.Hostname = hostname
+
+	return svcContainer
+}
+
+func containers() []ServiceContainer {
+	endpoint := "tcp://localhost:2375"
+	client, _ := docker.NewClient(endpoint)
+	containers, err := client.ListContainers(docker.ListContainersOptions{All: false})
+	if err != nil {
+		return nil
+	}
+
+	var containerList []ServiceContainer
+
+	for _, container := range containers {
+		containerList = append(containerList, toServiceContainer(container))
+	}
+
+	return containerList
+}

http.go

@@ -0,0 +1,42 @@
+package main
+
+import (
+	"net/http"
+
+	"github.com/gorilla/mux"
+	"github.com/hashicorp/memberlist"
+)
+
+
+func makeQueryHandler(fn func (http.ResponseWriter, *http.Request, *memberlist.Memberlist), list *memberlist.Memberlist) http.HandlerFunc {
+	return func(response http.ResponseWriter, req *http.Request) {
+		fn(response, req, list)
+	}
+}
+
+func servicesQueryHandler(response http.ResponseWriter, req *http.Request, list *memberlist.Memberlist) {
+	//params := mux.Vars(req)
+
+	defer req.Body.Close()
+
+	response.Header().Set("Content-Type", "text/html")
+	response.Write(
+		[]byte(`
+ 			<head>
+ 			<meta http-equiv="refresh" content="4">
+ 			</head> 
+	    	<pre>` + formatServices(list) + "</pre>"))
+}
+
+func serveHttp(list *memberlist.Memberlist) {
+	router := mux.NewRouter()
+
+	router.HandleFunc(
+		"/services", makeQueryHandler(servicesQueryHandler, list),
+	).Methods("GET")
+
+	http.Handle("/", router)
+
+	err := http.ListenAndServe("0.0.0.0:7777", nil)
+	exitWithError(err, "Can't start HTTP server")
+}

printing_delegate.go

@@ -0,0 +1,30 @@
+package main
+
+import (
+	"fmt"
+)
+
+type printingDelegate struct {}
+
+func (d *printingDelegate) NodeMeta(limit int) []byte {
+	fmt.Printf("NodeMeta(): %d\n", limit)
+	return []byte(`{ "State": "Running" }`)
+}
+
+func (d *printingDelegate) NotifyMsg(message []byte) {
+	fmt.Printf("NotifyMsg(): %s\n", string(message))
+}
+
+func (d *printingDelegate) GetBroadcasts(overhead, limit int) [][]byte {
+	fmt.Printf("GetBroadcasts(): %d %d\n", overhead, limit)
+	return nil
+}
+
+func (d *printingDelegate) LocalState(join bool) []byte {
+	fmt.Printf("LocalState(): %b\n", join)
+	return []byte("Some state")
+}
+
+func (d *printingDelegate) MergeRemoteState(buf []byte, join bool) {
+	fmt.Printf("MergeRemoteState(): %s %b\n", string(buf), join)
+}