Recovery should be < 500

* Augment the router penalty and decay as a hack fix

* increase recovery cost

api/handler/api/util_test.go

@@ -8,7 +8,7 @@ import (
 
 func TestRequestToProto(t *testing.T) {
 	testData := []*http.Request{
-		&http.Request{
+		{
 			Method: "GET",
 			Header: http.Header{
 				"Header": []string{"test"},

api/handler/http/http_test.go

@@ -27,7 +27,7 @@ func testHttp(t *testing.T, path, service, ns string) {
 	s := &registry.Service{
 		Name: service,
 		Nodes: []*registry.Node{
-			&registry.Node{
+			{
 				Id:      service + "-1",
 				Address: l.Addr().String(),
 			},

broker/common_test.go

@@ -7,7 +7,7 @@ import (
 var (
 	// mock data
 	testData = map[string][]*registry.Service{
-		"foo": []*registry.Service{
+		"foo": {
 			{
 				Name:    "foo",
 				Version: "1.0.0",

broker/http_broker_test.go

@@ -125,7 +125,7 @@ func pub(be *testing.B, c int) {
 
 	for i := 0; i < c; i++ {
 		go func() {
-			for _ = range ch {
+			for range ch {
 				if err := b.Publish(topic, msg); err != nil {
 					be.Fatalf("Unexpected publish error: %v", err)
 				}

client/common_test.go

@@ -7,7 +7,7 @@ import (
 var (
 	// mock data
 	testData = map[string][]*registry.Service{
-		"foo": []*registry.Service{
+		"foo": {
 			{
 				Name:    "foo",
 				Version: "1.0.0",

client/grpc/grpc_test.go

@@ -42,7 +42,7 @@ func TestGRPCClient(t *testing.T) {
 		Name:    "helloworld",
 		Version: "test",
 		Nodes: []*registry.Node{
-			&registry.Node{
+			{
 				Id:      "test-1",
 				Address: l.Addr().String(),
 			},

client/rpc_client_test.go

@@ -143,7 +143,7 @@ func TestCallWrapper(t *testing.T) {
 		Name:    service,
 		Version: "latest",
 		Nodes: []*registry.Node{
-			&registry.Node{
+			{
 				Id:      id,
 				Address: address,
 			},

client/selector/common_test.go

@@ -7,7 +7,7 @@ import (
 var (
 	// mock data
 	testData = map[string][]*registry.Service{
-		"foo": []*registry.Service{
+		"foo": {
 			{
 				Name:    "foo",
 				Version: "1.0.0",

client/selector/dns/dns.go

@@ -72,7 +72,7 @@ func (d *dnsSelector) Select(service string, opts ...selector.SelectOption) (sel
 	}
 
 	services := []*registry.Service{
-		&registry.Service{
+		{
 			Name:  service,
 			Nodes: nodes,
 		},

client/selector/filter_test.go

@@ -14,20 +14,20 @@ func TestFilterEndpoint(t *testing.T) {
 	}{
 		{
 			services: []*registry.Service{
-				&registry.Service{
+				{
 					Name:    "test",
 					Version: "1.0.0",
 					Endpoints: []*registry.Endpoint{
-						&registry.Endpoint{
+						{
 							Name: "Foo.Bar",
 						},
 					},
 				},
-				&registry.Service{
+				{
 					Name:    "test",
 					Version: "1.1.0",
 					Endpoints: []*registry.Endpoint{
-						&registry.Endpoint{
+						{
 							Name: "Baz.Bar",
 						},
 					},
@@ -38,20 +38,20 @@ func TestFilterEndpoint(t *testing.T) {
 		},
 		{
 			services: []*registry.Service{
-				&registry.Service{
+				{
 					Name:    "test",
 					Version: "1.0.0",
 					Endpoints: []*registry.Endpoint{
-						&registry.Endpoint{
+						{
 							Name: "Foo.Bar",
 						},
 					},
 				},
-				&registry.Service{
+				{
 					Name:    "test",
 					Version: "1.1.0",
 					Endpoints: []*registry.Endpoint{
-						&registry.Endpoint{
+						{
 							Name: "Foo.Bar",
 						},
 					},
@@ -95,11 +95,11 @@ func TestFilterLabel(t *testing.T) {
 	}{
 		{
 			services: []*registry.Service{
-				&registry.Service{
+				{
 					Name:    "test",
 					Version: "1.0.0",
 					Nodes: []*registry.Node{
-						&registry.Node{
+						{
 							Id:      "test-1",
 							Address: "localhost",
 							Metadata: map[string]string{
@@ -108,11 +108,11 @@ func TestFilterLabel(t *testing.T) {
 						},
 					},
 				},
-				&registry.Service{
+				{
 					Name:    "test",
 					Version: "1.1.0",
 					Nodes: []*registry.Node{
-						&registry.Node{
+						{
 							Id:      "test-2",
 							Address: "localhost",
 							Metadata: map[string]string{
@@ -127,21 +127,21 @@ func TestFilterLabel(t *testing.T) {
 		},
 		{
 			services: []*registry.Service{
-				&registry.Service{
+				{
 					Name:    "test",
 					Version: "1.0.0",
 					Nodes: []*registry.Node{
-						&registry.Node{
+						{
 							Id:      "test-1",
 							Address: "localhost",
 						},
 					},
 				},
-				&registry.Service{
+				{
 					Name:    "test",
 					Version: "1.1.0",
 					Nodes: []*registry.Node{
-						&registry.Node{
+						{
 							Id:      "test-2",
 							Address: "localhost",
 						},
@@ -187,11 +187,11 @@ func TestFilterVersion(t *testing.T) {
 	}{
 		{
 			services: []*registry.Service{
-				&registry.Service{
+				{
 					Name:    "test",
 					Version: "1.0.0",
 				},
-				&registry.Service{
+				{
 					Name:    "test",
 					Version: "1.1.0",
 				},
@@ -201,11 +201,11 @@ func TestFilterVersion(t *testing.T) {
 		},
 		{
 			services: []*registry.Service{
-				&registry.Service{
+				{
 					Name:    "test",
 					Version: "1.0.0",
 				},
-				&registry.Service{
+				{
 					Name:    "test",
 					Version: "1.1.0",
 				},

client/selector/strategy_test.go

@@ -8,29 +8,29 @@ import (
 
 func TestStrategies(t *testing.T) {
 	testData := []*registry.Service{
-		&registry.Service{
+		{
 			Name:    "test1",
 			Version: "latest",
 			Nodes: []*registry.Node{
-				&registry.Node{
+				{
 					Id:      "test1-1",
 					Address: "10.0.0.1:1001",
 				},
-				&registry.Node{
+				{
 					Id:      "test1-2",
 					Address: "10.0.0.2:1002",
 				},
 			},
 		},
-		&registry.Service{
+		{
 			Name:    "test1",
 			Version: "default",
 			Nodes: []*registry.Node{
-				&registry.Node{
+				{
 					Id:      "test1-3",
 					Address: "10.0.0.3:1003",
 				},
-				&registry.Node{
+				{
 					Id:      "test1-4",
 					Address: "10.0.0.4:1004",
 				},

common_test.go

@@ -7,7 +7,7 @@ import (
 var (
 	// mock data
 	testData = map[string][]*registry.Service{
-		"foo": []*registry.Service{
+		"foo": {
 			{
 				Name:    "foo",
 				Version: "1.0.0",

config/cmd/cmd.go

@@ -44,6 +44,10 @@ import (
 	thttp "github.com/micro/go-micro/transport/http"
 	tmem "github.com/micro/go-micro/transport/memory"
 	"github.com/micro/go-micro/transport/quic"
+
+	// runtimes
+	"github.com/micro/go-micro/runtime"
+	"github.com/micro/go-micro/runtime/kubernetes"
 )
 
 type Cmd interface {
@@ -67,6 +71,12 @@ var (
 	DefaultCmd = newCmd()
 
 	DefaultFlags = []cli.Flag{
+		cli.StringFlag{
+			Name:   "runtime",
+			Usage:  "Micro runtime",
+			EnvVar: "MICRO_RUNTIME",
+			Value:  "local",
+		},
 		cli.StringFlag{
 			Name:   "client",
 			EnvVar: "MICRO_CLIENT",
@@ -221,6 +231,11 @@ var (
 		"quic":   quic.NewTransport,
 	}
 
+	DefaultRuntimes = map[string]func(...runtime.Option) runtime.Runtime{
+		"local":      runtime.NewRuntime,
+		"kubernetes": kubernetes.NewRuntime,
+	}
+
 	// used for default selection as the fall back
 	defaultClient    = "rpc"
 	defaultServer    = "rpc"
@@ -228,6 +243,7 @@ var (
 	defaultRegistry  = "mdns"
 	defaultSelector  = "registry"
 	defaultTransport = "http"
+	defaultRuntime   = "local"
 )
 
 func init() {
@@ -247,6 +263,7 @@ func newCmd(opts ...Option) Cmd {
 		Server:    &server.DefaultServer,
 		Selector:  &selector.DefaultSelector,
 		Transport: &transport.DefaultTransport,
+		Runtime:   &runtime.DefaultRuntime,
 
 		Brokers:    DefaultBrokers,
 		Clients:    DefaultClients,
@@ -254,6 +271,7 @@ func newCmd(opts ...Option) Cmd {
 		Selectors:  DefaultSelectors,
 		Servers:    DefaultServers,
 		Transports: DefaultTransports,
+		Runtimes:   DefaultRuntimes,
 	}
 
 	for _, o := range opts {
@@ -294,6 +312,16 @@ func (c *cmd) Before(ctx *cli.Context) error {
 	var serverOpts []server.Option
 	var clientOpts []client.Option
 
+	// Set the runtime
+	if name := ctx.String("runtime"); len(name) > 0 {
+		r, ok := c.opts.Runtimes[name]
+		if !ok {
+			return fmt.Errorf("Unsupported runtime: %s", name)
+		}
+
+		*c.opts.Runtime = r()
+	}
+
 	// Set the client
 	if name := ctx.String("client"); len(name) > 0 {
 		// only change if we have the client and type differs

config/cmd/options.go

@@ -7,6 +7,7 @@ import (
 	"github.com/micro/go-micro/client"
 	"github.com/micro/go-micro/client/selector"
 	"github.com/micro/go-micro/registry"
+	"github.com/micro/go-micro/runtime"
 	"github.com/micro/go-micro/server"
 	"github.com/micro/go-micro/transport"
 )
@@ -24,6 +25,7 @@ type Options struct {
 	Transport *transport.Transport
 	Client    *client.Client
 	Server    *server.Server
+	Runtime   *runtime.Runtime
 
 	Brokers    map[string]func(...broker.Option) broker.Broker
 	Clients    map[string]func(...client.Option) client.Client
@@ -31,6 +33,7 @@ type Options struct {
 	Selectors  map[string]func(...selector.Option) selector.Selector
 	Servers    map[string]func(...server.Option) server.Server
 	Transports map[string]func(...transport.Option) transport.Transport
+	Runtimes   map[string]func(...runtime.Option) runtime.Runtime
 
 	// Other options for implementations of the interface
 	// can be stored in a context
@@ -135,3 +138,10 @@ func NewTransport(name string, t func(...transport.Option) transport.Transport)
 		o.Transports[name] = t
 	}
 }
+
+// New runtime func
+func NewRuntime(name string, r func(...runtime.Option) runtime.Runtime) Option {
+	return func(o *Options) {
+		o.Runtimes[name] = r
+	}
+}

config/reader/json/values_test.go

@@ -62,7 +62,7 @@ func TestStructArray(t *testing.T) {
 		{
 			[]byte(`[{"foo": "bar"}]`),
 			emptyTSlice,
-			[]T{T{Foo: "bar"}},
+			[]T{{Foo: "bar"}},
 		},
 	}
 

errors/errors_test.go

@@ -7,7 +7,7 @@ import (
 
 func TestErrors(t *testing.T) {
 	testData := []*Error{
-		&Error{
+		{
 			Id:     "test",
 			Code:   500,
 			Detail: "Internal server error",

go.mod

@@ -34,7 +34,9 @@ require (
 	github.com/jonboulle/clockwork v0.1.0 // indirect
 	github.com/joncalhoun/qson v0.0.0-20170526102502-8a9cab3a62b1
 	github.com/json-iterator/go v1.1.7
+	github.com/kr/pretty v0.1.0
 	github.com/leodido/go-urn v1.1.0 // indirect
+	github.com/lib/pq v1.2.0
 	github.com/lucas-clemente/quic-go v0.12.1
 	github.com/mholt/certmagic v0.7.5
 	github.com/micro/cli v0.2.0

go.sum

@@ -228,6 +228,8 @@ github.com/labbsr0x/bindman-dns-webhook v1.0.2/go.mod h1:p6b+VCXIR8NYKpDr8/dg1HK
 github.com/labbsr0x/goh v1.0.1/go.mod h1:8K2UhVoaWXcCU7Lxoa2omWnC8gyW8px7/lmO61c027w=
 github.com/leodido/go-urn v1.1.0 h1:Sm1gr51B1kKyfD2BlRcLSiEkffoG96g6TPv6eRoEiB8=
 github.com/leodido/go-urn v1.1.0/go.mod h1:+cyI34gQWZcE1eQU7NVgKkkzdXDQHr1dBMtdAPozLkw=
+github.com/lib/pq v1.2.0 h1:LXpIM/LZ5xGFhOpXAQUIMM1HdyqzVYM13zNdjCEEcA0=
+github.com/lib/pq v1.2.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
 github.com/linode/linodego v0.10.0/go.mod h1:cziNP7pbvE3mXIPneHj0oRY8L1WtGEIKlZ8LANE4eXA=
 github.com/liquidweb/liquidweb-go v1.6.0/go.mod h1:UDcVnAMDkZxpw4Y7NOHkqoeiGacVLEIG/i5J9cyixzQ=
 github.com/lucas-clemente/quic-go v0.12.1 h1:BPITli+6KnKogtTxBk2aS4okr5dUHz2LtIDAP1b8UL4=

metadata/metadata.go

@@ -12,6 +12,7 @@ type metaKey struct{}
 // from Transport headers.
 type Metadata map[string]string
 
+// Copy makes a copy of the metadata
 func Copy(md Metadata) Metadata {
 	cmd := make(Metadata)
 	for k, v := range md {
@@ -20,11 +21,31 @@ func Copy(md Metadata) Metadata {
 	return cmd
 }
 
+// FromContext returns metadata from the given context
 func FromContext(ctx context.Context) (Metadata, bool) {
 	md, ok := ctx.Value(metaKey{}).(Metadata)
 	return md, ok
 }
 
+// NewContext creates a new context with the given metadata
 func NewContext(ctx context.Context, md Metadata) context.Context {
 	return context.WithValue(ctx, metaKey{}, md)
 }
+
+// MergeContext merges metadata to existing metadata, overwriting if specified
+func MergeContext(ctx context.Context, patchMd Metadata, overwrite bool) context.Context {
+	md, _ := ctx.Value(metaKey{}).(Metadata)
+	cmd := make(Metadata)
+	for k, v := range md {
+		cmd[k] = v
+	}
+	for k, v := range patchMd {
+		if _, ok := cmd[k]; ok && !overwrite {
+			// skip
+		} else {
+			cmd[k] = v
+		}
+	}
+	return context.WithValue(ctx, metaKey{}, cmd)
+
+}

metadata/metadata_test.go

@@ -2,6 +2,7 @@ package metadata
 
 import (
 	"context"
+	"reflect"
 	"testing"
 )
 
@@ -40,3 +41,42 @@ func TestMetadataContext(t *testing.T) {
 		t.Errorf("Expected metadata length 1 got %d", i)
 	}
 }
+
+func TestMergeContext(t *testing.T) {
+	type args struct {
+		existing  Metadata
+		append    Metadata
+		overwrite bool
+	}
+	tests := []struct {
+		name string
+		args args
+		want Metadata
+	}{
+		{
+			name: "matching key, overwrite false",
+			args: args{
+				existing:  Metadata{"foo": "bar", "sumo": "demo"},
+				append:    Metadata{"sumo": "demo2"},
+				overwrite: false,
+			},
+			want: Metadata{"foo": "bar", "sumo": "demo"},
+		},
+		{
+			name: "matching key, overwrite true",
+			args: args{
+				existing:  Metadata{"foo": "bar", "sumo": "demo"},
+				append:    Metadata{"sumo": "demo2"},
+				overwrite: true,
+			},
+			want: Metadata{"foo": "bar", "sumo": "demo2"},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got, _ := FromContext(MergeContext(NewContext(context.TODO(), tt.args.existing), tt.args.append, tt.args.overwrite)); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("MergeContext() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

monitor/default.go

@@ -78,13 +78,6 @@ func (m *monitor) check(service string) (*Status, error) {
 				client.WithRetries(3),
 			)
 			if err != nil {
-				// reap the dead node
-				m.registry.Deregister(&registry.Service{
-					Name:    service.Name,
-					Version: service.Version,
-					Nodes:   []*registry.Node{node},
-				})
-
 				// save the error
 				gerr = err
 				continue
@@ -140,7 +133,7 @@ func (m *monitor) reap() {
 	defer m.Unlock()
 
 	// range over our watched services
-	for service, _ := range m.services {
+	for service := range m.services {
 		// check if the service exists in the registry
 		if !serviceMap[service] {
 			// if not, delete it in our status map
@@ -195,14 +188,14 @@ func (m *monitor) run() {
 			serviceMap := make(map[string]bool)
 
 			m.RLock()
-			for service, _ := range m.services {
+			for service := range m.services {
 				serviceMap[service] = true
 			}
 			m.RUnlock()
 
 			go func() {
 				// check the status of all watched services
-				for service, _ := range serviceMap {
+				for service := range serviceMap {
 					select {
 					case <-m.exit:
 						return
@@ -307,7 +300,7 @@ func (m *monitor) Stop() error {
 		return nil
 	default:
 		close(m.exit)
-		for s, _ := range m.services {
+		for s := range m.services {
 			delete(m.services, s)
 		}
 		m.registry.Stop()

network/default.go

@@ -79,21 +79,6 @@ func newNetwork(opts ...Option) Network {
 		o(&options)
 	}
 
-	// init tunnel address to the network bind address
-	options.Tunnel.Init(
-		tunnel.Address(options.Address),
-	)
-
-	// init router Id to the network id
-	options.Router.Init(
-		router.Id(options.Id),
-	)
-
-	// create tunnel client with tunnel transport
-	tunTransport := tun.NewTransport(
-		tun.WithTunnel(options.Tunnel),
-	)
-
 	// set the address to a hashed address
 	hasher := fnv.New64()
 	hasher.Write([]byte(options.Address + options.Id))
@@ -111,6 +96,22 @@ func newNetwork(opts ...Option) Network {
 		peerAddress = address
 	}
 
+	// init tunnel address to the network bind address
+	options.Tunnel.Init(
+		tunnel.Address(options.Address),
+	)
+
+	// init router Id to the network id
+	options.Router.Init(
+		router.Id(options.Id),
+		router.Address(peerAddress),
+	)
+
+	// create tunnel client with tunnel transport
+	tunTransport := tun.NewTransport(
+		tun.WithTunnel(options.Tunnel),
+	)
+
 	// server is network server
 	server := server.NewServer(
 		server.Id(options.Id),
@@ -308,7 +309,7 @@ func (n *network) acceptNetConn(l tunnel.Listener, recv chan *message) {
 func (n *network) updatePeerLinks(peerAddr string, linkId string) error {
 	n.Lock()
 	defer n.Unlock()
-	log.Debugf("Network looking up link %s in the peer links", linkId)
+	log.Tracef("Network looking up link %s in the peer links", linkId)
 	// lookup the peer link
 	var peerLink tunnel.Link
 	for _, link := range n.tunnel.Links() {
@@ -321,7 +322,7 @@ func (n *network) updatePeerLinks(peerAddr string, linkId string) error {
 		return ErrPeerLinkNotFound
 	}
 	// if the peerLink is found in the returned links update peerLinks
-	log.Debugf("Network updating peer links for peer %s", peerAddr)
+	log.Tracef("Network updating peer links for peer %s", peerAddr)
 	// add peerLink to the peerLinks map
 	if link, ok := n.peerLinks[peerAddr]; ok {
 		// if the existing has better Length then the new, replace it
@@ -368,7 +369,7 @@ func (n *network) processNetChan(listener tunnel.Listener) {
 					lastSeen: now,
 				}
 				// update peer links
-				log.Debugf("Network updating peer link %s for peer: %s", m.session.Link(), pbNetConnect.Node.Address)
+				log.Tracef("Network updating peer link %s for peer: %s", m.session.Link(), pbNetConnect.Node.Address)
 				if err := n.updatePeerLinks(pbNetConnect.Node.Address, m.session.Link()); err != nil {
 					log.Debugf("Network failed updating peer links: %s", err)
 				}
@@ -411,7 +412,7 @@ func (n *network) processNetChan(listener tunnel.Listener) {
 					lastSeen: now,
 				}
 				// update peer links
-				log.Debugf("Network updating peer link %s for peer: %s", m.session.Link(), pbNetPeer.Node.Address)
+				log.Tracef("Network updating peer link %s for peer: %s", m.session.Link(), pbNetPeer.Node.Address)
 				if err := n.updatePeerLinks(pbNetPeer.Node.Address, m.session.Link()); err != nil {
 					log.Debugf("Network failed updating peer links: %s", err)
 				}
@@ -438,7 +439,7 @@ func (n *network) processNetChan(listener tunnel.Listener) {
 
 				// NOTE: we don't unpack MaxDepth toplogy
 				peer = UnpackPeerTopology(pbNetPeer, now, MaxDepth-1)
-				log.Debugf("Network updating topology of node: %s", n.node.id)
+				log.Tracef("Network updating topology of node: %s", n.node.id)
 				if err := n.node.UpdatePeer(peer); err != nil {
 					log.Debugf("Network failed to update peers: %v", err)
 				}
@@ -700,16 +701,14 @@ func (n *network) getRouteMetric(router string, gateway string, link string) int
 	defer n.RUnlock()
 
 	if link == "local" && gateway == "" {
-		log.Debugf("Network link: %s, gateway: blank", link)
 		return 1
 	}
 
 	if link == "local" && gateway != "" {
-		log.Debugf("Network link: %s, gateway: %s", link, gateway)
 		return 2
 	}
 
-	log.Debugf("Network looking up %s link to gateway: %s", link, gateway)
+	log.Tracef("Network looking up %s link to gateway: %s", link, gateway)
 	if link, ok := n.peerLinks[gateway]; ok {
 		// maka sure delay is non-zero
 		delay := link.Delay()
@@ -724,7 +723,7 @@ func (n *network) getRouteMetric(router string, gateway string, link string) int
 			log.Debugf("Link length is 0 %v %v", link, link.Length())
 			length = 10e9
 		}
-		log.Debugf("Network calculated metric %v delay %v length %v distance %v", (delay*length*int64(hops))/10e6, delay, length, hops)
+		log.Tracef("Network calculated metric %v delay %v length %v distance %v", (delay*length*int64(hops))/10e6, delay, length, hops)
 		return (delay * length * int64(hops)) / 10e6
 	}
 
@@ -788,7 +787,7 @@ func (n *network) processCtrlChan(listener tunnel.Listener) {
 					// calculate route metric and add to the advertised metric
 					// we need to make sure we do not overflow math.MaxInt64
 					metric := n.getRouteMetric(event.Route.Router, event.Route.Gateway, event.Route.Link)
-					log.Debugf("Network metric for router %s and gateway %s: %v", event.Route.Router, event.Route.Gateway, metric)
+					log.Tracef("Network metric for router %s and gateway %s: %v", event.Route.Router, event.Route.Gateway, metric)
 
 					// check we don't overflow max int 64
 					if d := route.Metric + metric; d > math.MaxInt64 || d <= 0 {
@@ -809,7 +808,7 @@ func (n *network) processCtrlChan(listener tunnel.Listener) {
 				}
 				// if no events are eligible for processing continue
 				if len(events) == 0 {
-					log.Debugf("Network no events to be processed by router: %s", n.options.Id)
+					log.Tracef("Network no events to be processed by router: %s", n.options.Id)
 					continue
 				}
 				// create an advert and process it
@@ -865,7 +864,10 @@ func (n *network) advertise(advertChan <-chan *router.Advert) {
 				if event.Route.Router == advert.Id {
 					// hash the service before advertising it
 					hasher.Reset()
-					hasher.Write([]byte(event.Route.Address + n.node.id))
+					// routes for multiple instances of a service will be collapsed here.
+					// TODO: once we store labels in the table this may need to change
+					// to include the labels in case they differ but highly unlikely
+					hasher.Write([]byte(event.Route.Service + n.node.Address()))
 					address = fmt.Sprintf("%d", hasher.Sum64())
 				}
 				// calculate route metric to advertise

network/resolver/resolver.go

@@ -11,5 +11,6 @@ type Resolver interface {
 
 // A resolved record
 type Record struct {
-	Address string `json:"address"`
+	Address  string `json:"address"`
+	Priority int64  `json:"priority"`
 }

plugin/default.go

@@ -121,5 +121,7 @@ func (p *plugin) Build(path string, c *Config) error {
 		return fmt.Errorf("Failed to create dir %s: %v", filepath.Dir(path), err)
 	}
 	cmd := exec.Command("go", "build", "-buildmode=plugin", "-o", path+".so", goFile)
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
 	return cmd.Run()
 }

proxy/http/http.go

@@ -113,7 +113,7 @@ func (p *Proxy) ServeRequest(ctx context.Context, req server.Request, rsp server
 
 		// set response headers
 		hdr = map[string]string{}
-		for k, _ := range hrsp.Header {
+		for k := range hrsp.Header {
 			hdr[k] = hrsp.Header.Get(k)
 		}
 		// write the header

proxy/mucp/mucp.go

@@ -16,6 +16,7 @@ import (
 	"github.com/micro/go-micro/codec/bytes"
 	"github.com/micro/go-micro/config/options"
 	"github.com/micro/go-micro/errors"
+	"github.com/micro/go-micro/metadata"
 	"github.com/micro/go-micro/proxy"
 	"github.com/micro/go-micro/router"
 	"github.com/micro/go-micro/server"
@@ -43,9 +44,6 @@ type Proxy struct {
 	// A fib of routes service:address
 	sync.RWMutex
 	Routes map[string]map[uint64]router.Route
-
-	// The channel to monitor watcher errors
-	errChan chan error
 }
 
 // read client request and write to server
@@ -96,6 +94,75 @@ func toNodes(routes []router.Route) []string {
 	return nodes
 }
 
+func toSlice(r map[uint64]router.Route) []router.Route {
+	var routes []router.Route
+	for _, v := range r {
+		routes = append(routes, v)
+	}
+
+	// sort the routes in order of metric
+	sort.Slice(routes, func(i, j int) bool { return routes[i].Metric < routes[j].Metric })
+
+	return routes
+}
+
+func (p *Proxy) filterRoutes(ctx context.Context, routes []router.Route) []router.Route {
+	md, ok := metadata.FromContext(ctx)
+	if !ok {
+		return routes
+	}
+
+	var filteredRoutes []router.Route
+
+	// filter the routes based on our headers
+	for _, route := range routes {
+		// process only routes for this id
+		if id := md["Micro-Router"]; len(id) > 0 {
+			if route.Router != id {
+				// skip routes that don't mwatch
+				continue
+			}
+		}
+
+		// only process routes with this network
+		if net := md["Micro-Network"]; len(net) > 0 {
+			if route.Network != net {
+				// skip routes that don't mwatch
+				continue
+			}
+		}
+
+		// process only this gateway
+		if gw := md["Micro-Gateway"]; len(gw) > 0 {
+			// if the gateway matches our address
+			// special case, take the routes with no gateway
+			// TODO: should we strip the gateway from the context?
+			if gw == p.Router.Options().Address {
+				if len(route.Gateway) > 0 && route.Gateway != gw {
+					continue
+				}
+				// otherwise its a local route and we're keeping it
+			} else {
+				// gateway does not match our own
+				if route.Gateway != gw {
+					continue
+				}
+			}
+		}
+
+		// TODO: address based filtering
+		// address := md["Micro-Address"]
+
+		// TODO: label based filtering
+		// requires new field in routing table : route.Labels
+
+		// passed the filter checks
+		filteredRoutes = append(filteredRoutes, route)
+	}
+
+	return filteredRoutes
+}
+
 func (p *Proxy) getLink(r router.Route) (client.Client, error) {
 	if r.Link == "local" || len(p.Links) == 0 {
 		return p.Client, nil
@@ -107,28 +174,27 @@ func (p *Proxy) getLink(r router.Route) (client.Client, error) {
 	return l, nil
 }
 
-func (p *Proxy) getRoute(service string) ([]router.Route, error) {
-	toSlice := func(r map[uint64]router.Route) []router.Route {
-		var routes []router.Route
-		for _, v := range r {
-			routes = append(routes, v)
-		}
-
-		// sort the routes in order of metric
-		sort.Slice(routes, func(i, j int) bool { return routes[i].Metric < routes[j].Metric })
-
-		return routes
-	}
-
+func (p *Proxy) getRoute(ctx context.Context, service string) ([]router.Route, error) {
 	// lookup the route cache first
 	p.Lock()
-	routes, ok := p.Routes[service]
+	cached, ok := p.Routes[service]
 	if ok {
 		p.Unlock()
-		return toSlice(routes), nil
+		routes := toSlice(cached)
+		return p.filterRoutes(ctx, routes), nil
 	}
 	p.Unlock()
 
+	// cache routes for the service
+	routes, err := p.cacheRoutes(service)
+	if err != nil {
+		return nil, err
+	}
+
+	return p.filterRoutes(ctx, routes), nil
+}
+
+func (p *Proxy) cacheRoutes(service string) ([]router.Route, error) {
 	// lookup the routes in the router
 	results, err := p.Router.Lookup(router.QueryService(service))
 	if err != nil {
@@ -149,18 +215,40 @@ func (p *Proxy) getRoute(service string) ([]router.Route, error) {
 		}
 		p.Routes[service][route.Hash()] = route
 	}
-	routes = p.Routes[service]
+	routes := p.Routes[service]
 	p.Unlock()
 
 	return toSlice(routes), nil
 }
 
-// manageRouteCache applies action on a given route to Proxy route cache
-func (p *Proxy) manageRouteCache(route router.Route, action string) error {
+// refreshMetrics will refresh any metrics for our local cached routes.
+// we may not receive new watch events for these as they change.
+func (p *Proxy) refreshMetrics() {
+	var services []string
+
+	// get a list of services to update
+	p.RLock()
+	for service := range p.Routes {
+		services = append(services, service)
+	}
+	p.RUnlock()
+
+	// get and cache the routes for the service
+	for _, service := range services {
+		p.cacheRoutes(service)
+	}
+}
+
+// manageRoutes applies action on a given route to Proxy route cache
+func (p *Proxy) manageRoutes(route router.Route, action string) error {
+	// we only cache what we are actually concerned with
+	p.Lock()
+	defer p.Unlock()
+
 	switch action {
 	case "create", "update":
 		if _, ok := p.Routes[route.Service]; !ok {
-			p.Routes[route.Service] = make(map[uint64]router.Route)
+			return fmt.Errorf("not called %s", route.Service)
 		}
 		p.Routes[route.Service][route.Hash()] = route
 	case "delete":
@@ -174,31 +262,22 @@ func (p *Proxy) manageRouteCache(route router.Route, action string) error {
 
 // watchRoutes watches service routes and updates proxy cache
 func (p *Proxy) watchRoutes() {
-	// this is safe to do as the only way watchRoutes returns is
-	// when some error is written into error channel - we want to bail then
-	defer close(p.errChan)
-
 	// route watcher
 	w, err := p.Router.Watch()
 	if err != nil {
-		p.errChan <- err
 		return
 	}
 
 	for {
 		event, err := w.Next()
 		if err != nil {
-			p.errChan <- err
 			return
 		}
 
-		p.Lock()
-		if err := p.manageRouteCache(event.Route, fmt.Sprintf("%s", event.Type)); err != nil {
+		if err := p.manageRoutes(event.Route, fmt.Sprintf("%s", event.Type)); err != nil {
 			// TODO: should we bail here?
-			p.Unlock()
 			continue
 		}
-		p.Unlock()
 	}
 }
 
@@ -235,7 +314,7 @@ func (p *Proxy) ServeRequest(ctx context.Context, req server.Request, rsp server
 			addresses = []string{p.Endpoint}
 		} else {
 			// get route for endpoint from router
-			addr, err := p.getRoute(p.Endpoint)
+			addr, err := p.getRoute(ctx, p.Endpoint)
 			if err != nil {
 				return err
 			}
@@ -247,7 +326,7 @@ func (p *Proxy) ServeRequest(ctx context.Context, req server.Request, rsp server
 	} else {
 		// no endpoint was specified just lookup the route
 		// get route for endpoint from router
-		addr, err := p.getRoute(service)
+		addr, err := p.getRoute(ctx, service)
 		if err != nil {
 			return err
 		}
@@ -360,39 +439,28 @@ func (p *Proxy) serveRequest(ctx context.Context, link client.Client, service, e
 	// get raw response
 	resp := stream.Response()
 
-	// route watcher error
-	var watchErr error
-
 	// create server response write loop
 	for {
-		select {
-		case err := <-p.errChan:
-			if err != nil {
-				watchErr = err
-			}
-			return watchErr
-		default:
-			// read backend response body
-			body, err := resp.Read()
-			if err == io.EOF {
-				return nil
-			} else if err != nil {
-				return err
-			}
+		// read backend response body
+		body, err := resp.Read()
+		if err == io.EOF {
+			return nil
+		} else if err != nil {
+			return err
+		}
 
-			// read backend response header
-			hdr := resp.Header()
+		// read backend response header
+		hdr := resp.Header()
 
-			// write raw response header to client
-			rsp.WriteHeader(hdr)
+		// write raw response header to client
+		rsp.WriteHeader(hdr)
 
-			// write raw response body to client
-			err = rsp.Write(body)
-			if err == io.EOF {
-				return nil
-			} else if err != nil {
-				return err
-			}
+		// write raw response body to client
+		err = rsp.Write(body)
+		if err == io.EOF {
+			return nil
+		} else if err != nil {
+			return err
 		}
 	}
 
@@ -451,9 +519,6 @@ func NewProxy(opts ...options.Option) proxy.Proxy {
 	// routes cache
 	p.Routes = make(map[string]map[uint64]router.Route)
 
-	// watch router service routes
-	p.errChan = make(chan error, 1)
-
 	go func() {
 		// continuously attempt to watch routes
 		for {
@@ -464,5 +529,19 @@ func NewProxy(opts ...options.Option) proxy.Proxy {
 		}
 	}()
 
+	go func() {
+		t := time.NewTicker(time.Minute)
+		defer t.Stop()
+
+		// we must refresh route metrics since they do not trigger new events
+		for {
+			select {
+			case <-t.C:
+				// refresh route metrics
+				p.refreshMetrics()
+			}
+		}
+	}()
+
 	return p
 }

registry/cache/cache.go

@@ -81,7 +81,7 @@ func (c *cache) isValid(services []*registry.Service, ttl time.Time) bool {
 	}
 
 	// time since ttl is longer than timeout
-	if time.Since(ttl) > c.opts.TTL {
+	if time.Since(ttl) > 0 {
 		return false
 	}
 

registry/encoding_test.go

@@ -6,13 +6,13 @@ import (
 
 func TestEncoding(t *testing.T) {
 	testData := []*mdnsTxt{
-		&mdnsTxt{
+		{
 			Version: "1.0.0",
 			Metadata: map[string]string{
 				"foo": "bar",
 			},
 			Endpoints: []*Endpoint{
-				&Endpoint{
+				{
 					Name: "endpoint1",
 					Request: &Value{
 						Name: "request",

registry/mdns_test.go

@@ -7,11 +7,11 @@ import (
 
 func TestMDNS(t *testing.T) {
 	testData := []*Service{
-		&Service{
+		{
 			Name:    "test1",
 			Version: "1.0.1",
 			Nodes: []*Node{
-				&Node{
+				{
 					Id:      "test1-1",
 					Address: "10.0.0.1:10001",
 					Metadata: map[string]string{
@@ -20,11 +20,11 @@ func TestMDNS(t *testing.T) {
 				},
 			},
 		},
-		&Service{
+		{
 			Name:    "test2",
 			Version: "1.0.2",
 			Nodes: []*Node{
-				&Node{
+				{
 					Id:      "test2-1",
 					Address: "10.0.0.2:10002",
 					Metadata: map[string]string{
@@ -33,11 +33,11 @@ func TestMDNS(t *testing.T) {
 				},
 			},
 		},
-		&Service{
+		{
 			Name:    "test3",
 			Version: "1.0.3",
 			Nodes: []*Node{
-				&Node{
+				{
 					Id:      "test3-1",
 					Address: "10.0.0.3:10003",
 					Metadata: map[string]string{

registry/memory/memory_test.go

@@ -8,7 +8,7 @@ import (
 
 var (
 	testData = map[string][]*registry.Service{
-		"foo": []*registry.Service{
+		"foo": {
 			{
 				Name:    "foo",
 				Version: "1.0.0",
@@ -44,7 +44,7 @@ var (
 				},
 			},
 		},
-		"bar": []*registry.Service{
+		"bar": {
 			{
 				Name:    "bar",
 				Version: "default",

registry/watcher_test.go

@@ -6,11 +6,11 @@ import (
 
 func TestWatcher(t *testing.T) {
 	testData := []*Service{
-		&Service{
+		{
 			Name:    "test1",
 			Version: "1.0.1",
 			Nodes: []*Node{
-				&Node{
+				{
 					Id:      "test1-1",
 					Address: "10.0.0.1:10001",
 					Metadata: map[string]string{
@@ -19,11 +19,11 @@ func TestWatcher(t *testing.T) {
 				},
 			},
 		},
-		&Service{
+		{
 			Name:    "test2",
 			Version: "1.0.2",
 			Nodes: []*Node{
-				&Node{
+				{
 					Id:      "test2-1",
 					Address: "10.0.0.2:10002",
 					Metadata: map[string]string{
@@ -32,11 +32,11 @@ func TestWatcher(t *testing.T) {
 				},
 			},
 		},
-		&Service{
+		{
 			Name:    "test3",
 			Version: "1.0.3",
 			Nodes: []*Node{
-				&Node{
+				{
 					Id:      "test3-1",
 					Address: "10.0.0.3:10003",
 					Metadata: map[string]string{

router/default.go

@@ -23,15 +23,13 @@ const (
 	// AdvertSuppress is advert suppression threshold
 	AdvertSuppress = 2000.0
 	// AdvertRecover is advert recovery threshold
-	AdvertRecover = 750.0
+	AdvertRecover = 500.0
 	// DefaultAdvertTTL is default advertisement TTL
 	DefaultAdvertTTL = 1 * time.Minute
-	// DeletePenalty penalises route deletion
-	DeletePenalty = 1000.0
-	// UpdatePenalty penalises route updates
-	UpdatePenalty = 500.0
+	// Penalty for routes processed multiple times
+	Penalty = 2000.0
 	// PenaltyHalfLife is the time the advert penalty decays to half its value
-	PenaltyHalfLife = 2.0
+	PenaltyHalfLife = 2.5
 	// MaxSuppressTime defines time after which the suppressed advert is deleted
 	MaxSuppressTime = 5 * time.Minute
 )
@@ -385,15 +383,16 @@ func (r *router) advertiseEvents() error {
 				// decay the event penalty
 				delta := time.Since(advert.lastUpdate).Seconds()
 				advert.penalty = advert.penalty * math.Exp(-delta*PenaltyDecay)
-
+				service := advert.event.Route.Service
+				address := advert.event.Route.Address
 				// suppress/recover the event based on its penalty level
 				switch {
 				case advert.penalty > AdvertSuppress && !advert.isSuppressed:
-					log.Debugf("Router supressing advert %d for route %s", key, advert.event.Route.Address)
+					log.Debugf("Router suppressing advert %d %.2f for route %s %s", key, advert.penalty, service, address)
 					advert.isSuppressed = true
 					advert.suppressTime = time.Now()
 				case advert.penalty < AdvertRecover && advert.isSuppressed:
-					log.Debugf("Router recovering advert %d for route %s", key, advert.event.Route.Address)
+					log.Debugf("Router recovering advert %d %.2f for route %s %s", key, advert.penalty, service, address)
 					advert.isSuppressed = false
 				}
 
@@ -425,15 +424,11 @@ func (r *router) advertiseEvents() error {
 			if e == nil {
 				continue
 			}
-			log.Debugf("Router processing table event %s for service %s", e.Type, e.Route.Address)
+			log.Debugf("Router processing table event %s for service %s %s", e.Type, e.Route.Service, e.Route.Address)
+
 			// determine the event penalty
-			var penalty float64
-			switch e.Type {
-			case Update:
-				penalty = UpdatePenalty
-			case Delete:
-				penalty = DeletePenalty
-			}
+			// TODO: should there be any difference in penalty for different event types
+			penalty := Penalty
 
 			// check if we have already registered the route
 			hash := e.Route.Hash()
@@ -456,7 +451,7 @@ func (r *router) advertiseEvents() error {
 			// update event penalty and timestamp
 			advert.lastUpdate = time.Now()
 			advert.penalty += penalty
-			log.Debugf("Router advert %d for route %s event penalty: %f", hash, advert.event.Route.Address, advert.penalty)
+			log.Debugf("Router advert %d for route %s %s event penalty: %f", hash, advert.event.Route.Service, advert.event.Route.Address, advert.penalty)
 		case <-r.exit:
 			// first wait for the advertiser to finish
 			r.advertWg.Wait()
@@ -665,7 +660,7 @@ func (r *router) Process(a *Advert) error {
 		// create a copy of the route
 		route := event.Route
 		action := event.Type
-		log.Debugf("Router %s applying %s from router %s for address: %s", r.options.Id, action, route.Router, route.Address)
+		log.Debugf("Router %s applying %s from router %s for service %s %s", r.options.Id, action, route.Router, route.Service, route.Address)
 		if err := r.manageRoute(route, fmt.Sprintf("%s", action)); err != nil {
 			return fmt.Errorf("failed applying action %s to routing table: %s", action, err)
 		}

router/service/service.go

@@ -230,7 +230,7 @@ func (s *svc) Solicit() error {
 
 	// build events to advertise
 	events := make([]*router.Event, len(routes))
-	for i, _ := range events {
+	for i := range events {
 		events[i] = &router.Event{
 			Type:      router.Update,
 			Timestamp: time.Now(),

runtime/default.go

@@ -2,19 +2,18 @@ package runtime
 
 import (
 	"errors"
-	"io"
-	"strings"
+	"fmt"
+	"strconv"
 	"sync"
 	"time"
 
-	"github.com/micro/go-micro/runtime/package"
-	"github.com/micro/go-micro/runtime/process"
-	proc "github.com/micro/go-micro/runtime/process/os"
 	"github.com/micro/go-micro/util/log"
 )
 
 type runtime struct {
 	sync.RWMutex
+	// options configure runtime
+	options Options
 	// used to stop the runtime
 	closed chan bool
 	// used to start new services
@@ -25,158 +24,38 @@ type runtime struct {
 	services map[string]*service
 }
 
-type service struct {
-	sync.RWMutex
+// NewRuntime creates new local runtime and returns it
+func NewRuntime(opts ...Option) Runtime {
+	// get default options
+	options := Options{}
 
-	running bool
-	closed  chan bool
-	err     error
+	// apply requested options
+	for _, o := range opts {
+		o(&options)
+	}
 
-	// output for logs
-	output io.Writer
-
-	// service to manage
-	*Service
-	// process creator
-	Process *proc.Process
-	// Exec
-	Exec *process.Executable
-	// process pid
-	PID *process.PID
-}
-
-func newRuntime() *runtime {
 	return &runtime{
+		options:  options,
 		closed:   make(chan bool),
 		start:    make(chan *service, 128),
 		services: make(map[string]*service),
 	}
 }
 
-func newService(s *Service, c CreateOptions) *service {
-	var exec string
-	var args []string
+// Init initializes runtime options
+func (r *runtime) Init(opts ...Option) error {
+	r.Lock()
+	defer r.Unlock()
 
-	if len(s.Exec) > 0 {
-		parts := strings.Split(s.Exec, " ")
-		exec = parts[0]
-		args = []string{}
-
-		if len(parts) > 1 {
-			args = parts[1:]
-		}
-	} else {
-		// set command
-		exec = c.Command[0]
-		// set args
-		if len(c.Command) > 1 {
-			args = c.Command[1:]
-		}
-	}
-
-	return &service{
-		Service: s,
-		Process: new(proc.Process),
-		Exec: &process.Executable{
-			Binary: &packager.Binary{
-				Name: s.Name,
-				Path: exec,
-			},
-			Env:  c.Env,
-			Args: args,
-		},
-		output: c.Output,
-	}
-}
-
-func (s *service) streamOutput() {
-	go io.Copy(s.output, s.PID.Output)
-	go io.Copy(s.output, s.PID.Error)
-}
-
-func (s *service) Running() bool {
-	s.RLock()
-	defer s.RUnlock()
-	return s.running
-}
-
-func (s *service) Start() error {
-	s.Lock()
-	defer s.Unlock()
-
-	if s.running {
-		return nil
-	}
-
-	// reset
-	s.err = nil
-	s.closed = make(chan bool)
-
-	// TODO: pull source & build binary
-	log.Debugf("Runtime service %s forking new process\n")
-	p, err := s.Process.Fork(s.Exec)
-	if err != nil {
-		return err
-	}
-
-	// set the pid
-	s.PID = p
-	// set to running
-	s.running = true
-
-	if s.output != nil {
-		s.streamOutput()
-	}
-
-	// wait and watch
-	go s.Wait()
-
-	return nil
-}
-
-func (s *service) Stop() error {
-	s.Lock()
-	defer s.Unlock()
-
-	select {
-	case <-s.closed:
-		return nil
-	default:
-		close(s.closed)
-		s.running = false
-		return s.Process.Kill(s.PID)
+	for _, o := range opts {
+		o(&r.options)
 	}
 
 	return nil
 }
 
-func (s *service) Error() error {
-	s.RLock()
-	defer s.RUnlock()
-	return s.err
-}
-
-func (s *service) Wait() {
-	// wait for process to exit
-	err := s.Process.Wait(s.PID)
-
-	s.Lock()
-	defer s.Unlock()
-
-	// save the error
-	if err != nil {
-		s.err = err
-	}
-
-	// no longer running
-	s.running = false
-}
-
-func (r *runtime) run() {
-	r.RLock()
-	closed := r.closed
-	r.RUnlock()
-
+// run runs the runtime management loop
+func (r *runtime) run(events <-chan Event) {
 	t := time.NewTicker(time.Second * 5)
 	defer t.Stop()
 
@@ -201,19 +80,67 @@ func (r *runtime) run() {
 			if service.Running() {
 				continue
 			}
-
 			// TODO: check service error
-			log.Debugf("Starting %s", service.Name)
+			log.Debugf("Runtime starting service %s", service.Name)
 			if err := service.Start(); err != nil {
-				log.Debugf("Runtime error starting %s: %v", service.Name, err)
+				log.Debugf("Runtime error starting service %s: %v", service.Name, err)
 			}
-		case <-closed:
-			// TODO: stop all the things
+		case event := <-events:
+			log.Debugf("Runtime received notification event: %v", event)
+			// NOTE: we only handle Update events for now
+			switch event.Type {
+			case Update:
+				// parse returned response to timestamp
+				updateTimeStamp, err := strconv.ParseInt(event.Version, 10, 64)
+				if err != nil {
+					log.Debugf("Runtime error parsing update build time: %v", err)
+					continue
+				}
+				buildTime := time.Unix(updateTimeStamp, 0)
+				processEvent := func(event Event, service *Service) error {
+					buildTimeStamp, err := strconv.ParseInt(service.Version, 10, 64)
+					if err != nil {
+						return err
+					}
+					muBuild := time.Unix(buildTimeStamp, 0)
+					if buildTime.After(muBuild) {
+						if err := r.Update(service); err != nil {
+							return err
+						}
+						service.Version = fmt.Sprintf("%d", buildTime.Unix())
+					}
+					return nil
+				}
+				r.Lock()
+				if len(event.Service) > 0 {
+					service, ok := r.services[event.Service]
+					if !ok {
+						log.Debugf("Runtime unknown service: %s", event.Service)
+						r.Unlock()
+						continue
+					}
+					if err := processEvent(event, service.Service); err != nil {
+						log.Debugf("Runtime error updating service %s: %v", event.Service, err)
+					}
+					r.Unlock()
+					continue
+				}
+				// if blank service was received we update all services
+				for _, service := range r.services {
+					if err := processEvent(event, service.Service); err != nil {
+						log.Debugf("Runtime error updating service %s: %v", service.Name, err)
+					}
+				}
+				r.Unlock()
+			}
+		case <-r.closed:
+			log.Debugf("Runtime stopped.")
 			return
 		}
 	}
 }
 
+// Create creates a new service which is then started by runtime
 func (r *runtime) Create(s *Service, opts ...CreateOption) error {
 	r.Lock()
 	defer r.Unlock()
@@ -240,6 +167,7 @@ func (r *runtime) Create(s *Service, opts ...CreateOption) error {
 	return nil
 }
 
+// Delete removes the service from the runtime and stops it
 func (r *runtime) Delete(s *Service) error {
 	r.Lock()
 	defer r.Unlock()
@@ -252,6 +180,31 @@ func (r *runtime) Delete(s *Service) error {
 	return nil
 }
 
+// Update attemps to update the service
+func (r *runtime) Update(s *Service) error {
+	// delete the service
+	if err := r.Delete(s); err != nil {
+		return err
+	}
+
+	// create new service
+	return r.Create(s)
+}
+
+// List returns a slice of all services tracked by the runtime
+func (r *runtime) List() ([]*Service, error) {
+	var services []*Service
+	r.RLock()
+	defer r.RUnlock()
+
+	for _, service := range r.services {
+		services = append(services, service.Service)
+	}
+
+	return services, nil
+}
+
+// Start starts the runtime
 func (r *runtime) Start() error {
 	r.Lock()
 	defer r.Unlock()
@@ -265,11 +218,22 @@ func (r *runtime) Start() error {
 	r.running = true
 	r.closed = make(chan bool)
 
-	go r.run()
+	var events <-chan Event
+	if r.options.Notifier != nil {
+		var err error
+		events, err = r.options.Notifier.Notify()
+		if err != nil {
+			// TODO: should we bail here?
+			log.Debugf("Runtime failed to start update notifier")
+		}
+	}
+
+	go r.run(events)
 
 	return nil
 }
 
+// Stop stops the runtime
 func (r *runtime) Stop() error {
 	r.Lock()
 	defer r.Unlock()
@@ -292,7 +256,16 @@ func (r *runtime) Stop() error {
 			log.Debugf("Runtime stopping %s", service.Name)
 			service.Stop()
 		}
+		// stop the notifier too
+		if r.options.Notifier != nil {
+			return r.options.Notifier.Close()
+		}
 	}
 
 	return nil
 }
+
+// String implements stringer interface
+func (r *runtime) String() string {
+	return "local"
+}

runtime/kubernetes/client/api/request.go

@@ -0,0 +1,223 @@
+package api
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+
+	"github.com/micro/go-micro/runtime/kubernetes/client/watch"
+	"github.com/micro/go-micro/util/log"
+)
+
+// Request is used to construct a http request for the k8s API.
+type Request struct {
+	client    *http.Client
+	header    http.Header
+	params    url.Values
+	method    string
+	host      string
+	namespace string
+
+	resource     string
+	resourceName *string
+	body         io.Reader
+
+	err error
+}
+
+// Params is the object to pass in to set paramaters
+// on a request.
+type Params struct {
+	LabelSelector map[string]string
+	Annotations   map[string]string
+	Watch         bool
+}
+
+// verb sets method
+func (r *Request) verb(method string) *Request {
+	r.method = method
+	return r
+}
+
+// Get request
+func (r *Request) Get() *Request {
+	return r.verb("GET")
+}
+
+// Post request
+func (r *Request) Post() *Request {
+	return r.verb("POST")
+}
+
+// Put request
+func (r *Request) Put() *Request {
+	return r.verb("PUT")
+}
+
+// Patch request
+// https://github.com/kubernetes/kubernetes/blob/master/docs/devel/api-conventions.md#patch-operations
+func (r *Request) Patch() *Request {
+	return r.verb("PATCH").SetHeader("Content-Type", "application/strategic-merge-patch+json")
+}
+
+// Delete request
+func (r *Request) Delete() *Request {
+	return r.verb("DELETE")
+}
+
+// Namespace is to set the namespace to operate on
+func (r *Request) Namespace(s string) *Request {
+	r.namespace = s
+	return r
+}
+
+// Resource is the type of resource the operation is
+// for, such as "services", "endpoints" or "pods"
+func (r *Request) Resource(s string) *Request {
+	r.resource = s
+	return r
+}
+
+// Name is for targeting a specific resource by id
+func (r *Request) Name(s string) *Request {
+	r.resourceName = &s
+	return r
+}
+
+// Body pass in a body to set, this is for POST, PUT
+// and PATCH requests
+func (r *Request) Body(in interface{}) *Request {
+	b := new(bytes.Buffer)
+	if err := json.NewEncoder(b).Encode(&in); err != nil {
+		r.err = err
+		return r
+	}
+	log.Debugf("Patch body: %v", b)
+	r.body = b
+	return r
+}
+
+// Params isused to set paramters on a request
+func (r *Request) Params(p *Params) *Request {
+	for k, v := range p.LabelSelector {
+		r.params.Add("labelSelectors", k+"="+v)
+	}
+
+	return r
+}
+
+// SetHeader sets a header on a request with
+// a `key` and `value`
+func (r *Request) SetHeader(key, value string) *Request {
+	r.header.Add(key, value)
+	return r
+}
+
+// request builds the http.Request from the options
+func (r *Request) request() (*http.Request, error) {
+	var url string
+	switch r.resource {
+	case "pods":
+		// /api/v1/namespaces/{namespace}/pods
+		url = fmt.Sprintf("%s/api/v1/namespaces/%s/%s/", r.host, r.namespace, r.resource)
+	case "deployments":
+		// /apis/apps/v1/namespaces/{namespace}/deployments/{name}
+		url = fmt.Sprintf("%s/apis/apps/v1/namespaces/%s/%s/", r.host, r.namespace, r.resource)
+	}
+
+	// append resourceName if it is present
+	if r.resourceName != nil {
+		url += *r.resourceName
+	}
+
+	// append any query params
+	if len(r.params) > 0 {
+		url += "?" + r.params.Encode()
+	}
+
+	// build request
+	req, err := http.NewRequest(r.method, url, r.body)
+	if err != nil {
+		return nil, err
+	}
+
+	// set headers on request
+	req.Header = r.header
+	return req, nil
+}
+
+// Do builds and triggers the request
+func (r *Request) Do() *Response {
+	if r.err != nil {
+		return &Response{
+			err: r.err,
+		}
+	}
+
+	req, err := r.request()
+	if err != nil {
+		return &Response{
+			err: err,
+		}
+	}
+
+	log.Debugf("kubernetes api request: %v", req)
+
+	res, err := r.client.Do(req)
+	if err != nil {
+		return &Response{
+			err: err,
+		}
+	}
+
+	log.Debugf("kubernetes api response: %v", res)
+
+	// return res, err
+	return newResponse(res, err)
+}
+
+// Watch builds and triggers the request, but
+// will watch instead of return an object
+func (r *Request) Watch() (watch.Watch, error) {
+	if r.err != nil {
+		return nil, r.err
+	}
+
+	r.params.Set("watch", "true")
+
+	req, err := r.request()
+	if err != nil {
+		return nil, err
+	}
+
+	w, err := watch.NewBodyWatcher(req, r.client)
+	return w, err
+}
+
+// Options ...
+type Options struct {
+	Host        string
+	Namespace   string
+	BearerToken *string
+	Client      *http.Client
+}
+
+// NewRequest creates a k8s api request
+func NewRequest(opts *Options) *Request {
+	req := &Request{
+		header:    make(http.Header),
+		params:    make(url.Values),
+		client:    opts.Client,
+		namespace: opts.Namespace,
+		host:      opts.Host,
+	}
+
+	if opts.BearerToken != nil {
+		req.SetHeader("Authorization", "Bearer "+*opts.BearerToken)
+	}
+
+	return req
+}

runtime/kubernetes/client/api/response.go

@@ -0,0 +1,94 @@
+package api
+
+import (
+	"encoding/json"
+	"errors"
+	"io/ioutil"
+	"net/http"
+
+	"github.com/micro/go-micro/util/log"
+)
+
+// Errors ...
+var (
+	ErrNotFound = errors.New("kubernetes: not found")
+	ErrDecode   = errors.New("kubernetes: error decoding")
+	ErrOther    = errors.New("kubernetes: unknown error")
+)
+
+// Status is an object that is returned when a request
+// failed or delete succeeded.
+// type Status struct {
+// 	Kind    string `json:"kind"`
+// 	Status  string `json:"status"`
+// 	Message string `json:"message"`
+// 	Reason  string `json:"reason"`
+// 	Code    int    `json:"code"`
+// }
+
+// Response ...
+type Response struct {
+	res *http.Response
+	err error
+
+	body []byte
+}
+
+// Error returns an error
+func (r *Response) Error() error {
+	return r.err
+}
+
+// StatusCode returns status code for response
+func (r *Response) StatusCode() int {
+	return r.res.StatusCode
+}
+
+// Into decode body into `data`
+func (r *Response) Into(data interface{}) error {
+	if r.err != nil {
+		return r.err
+	}
+
+	defer r.res.Body.Close()
+	decoder := json.NewDecoder(r.res.Body)
+	err := decoder.Decode(&data)
+	if err != nil {
+		return ErrDecode
+	}
+
+	return r.err
+}
+
+func newResponse(res *http.Response, err error) *Response {
+	r := &Response{
+		res: res,
+		err: err,
+	}
+
+	if err != nil {
+		return r
+	}
+
+	if r.res.StatusCode == http.StatusOK ||
+		r.res.StatusCode == http.StatusCreated ||
+		r.res.StatusCode == http.StatusNoContent {
+		// Non error status code
+		return r
+	}
+
+	if r.res.StatusCode == http.StatusNotFound {
+		r.err = ErrNotFound
+		return r
+	}
+
+	log.Logf("kubernetes: request failed with code %v", r.res.StatusCode)
+
+	b, err := ioutil.ReadAll(r.res.Body)
+	if err == nil {
+		log.Log("kubernetes: request failed with body:")
+		log.Log(string(b))
+	}
+	r.err = ErrOther
+	return r
+}

runtime/kubernetes/client/client.go

@@ -0,0 +1,102 @@
+package client
+
+import (
+	"crypto/tls"
+	"errors"
+	"io/ioutil"
+	"net/http"
+	"os"
+	"path"
+
+	"github.com/micro/go-micro/runtime/kubernetes/client/api"
+	"github.com/micro/go-micro/util/log"
+)
+
+var (
+	serviceAccountPath = "/var/run/secrets/kubernetes.io/serviceaccount"
+	// ErrReadNamespace is returned when the names could not be read from service account
+	ErrReadNamespace = errors.New("Could not read namespace from service account secret")
+)
+
+// Client ...
+type client struct {
+	opts *api.Options
+}
+
+// NewClientInCluster should work similarily to the official api
+// NewInClient by setting up a client configuration for use within
+// a k8s pod.
+func NewClientInCluster() *client {
+	host := "https://" + os.Getenv("KUBERNETES_SERVICE_HOST") + ":" + os.Getenv("KUBERNETES_SERVICE_PORT")
+
+	s, err := os.Stat(serviceAccountPath)
+	if err != nil {
+		log.Fatal(err)
+	}
+	if s == nil || !s.IsDir() {
+		log.Fatal(errors.New("no k8s service account found"))
+	}
+
+	token, err := ioutil.ReadFile(path.Join(serviceAccountPath, "token"))
+	if err != nil {
+		log.Fatal(err)
+	}
+	t := string(token)
+
+	ns, err := detectNamespace()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	crt, err := CertPoolFromFile(path.Join(serviceAccountPath, "ca.crt"))
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	c := &http.Client{
+		Transport: &http.Transport{
+			TLSClientConfig: &tls.Config{
+				RootCAs: crt,
+			},
+			DisableCompression: true,
+		},
+	}
+
+	return &client{
+		opts: &api.Options{
+			Client:      c,
+			Host:        host,
+			Namespace:   ns,
+			BearerToken: &t,
+		},
+	}
+}
+
+func detectNamespace() (string, error) {
+	nsPath := path.Join(serviceAccountPath, "namespace")
+
+	// Make sure it's a file and we can read it
+	if s, e := os.Stat(nsPath); e != nil {
+		return "", e
+	} else if s.IsDir() {
+		return "", ErrReadNamespace
+	}
+
+	// Read the file, and cast to a string
+	if ns, e := ioutil.ReadFile(nsPath); e != nil {
+		return string(ns), e
+	} else {
+		return string(ns), nil
+	}
+}
+
+// UpdateDeployment
+func (c *client) UpdateDeployment(name string, body interface{}) error {
+	return api.NewRequest(c.opts).
+		Patch().
+		Resource("deployments").
+		Name(name).
+		Body(body).
+		Do().
+		Error()
+}

runtime/kubernetes/client/kubernetes.go

@@ -0,0 +1,12 @@
+package client
+
+// Kubernetes client
+type Kubernetes interface {
+	// UpdateDeployment patches deployment annotations with new metadata
+	UpdateDeployment(string, interface{}) error
+}
+
+// Metadata defines api request metadata
+type Metadata struct {
+	Annotations map[string]string `json:"annotations,omitempty"`
+}

runtime/kubernetes/client/utils.go

@@ -0,0 +1,74 @@
+package client
+
+import (
+	"crypto/x509"
+	"encoding/pem"
+	"errors"
+	"fmt"
+	"io/ioutil"
+)
+
+// COPIED FROM
+// https://github.com/kubernetes/kubernetes/blob/7a725418af4661067b56506faabc2d44c6d7703a/pkg/util/crypto/crypto.go
+
+// CertPoolFromFile returns an x509.CertPool containing the certificates in the given PEM-encoded file.
+// Returns an error if the file could not be read, a certificate could not be parsed, or if the file does not contain any certificates
+func CertPoolFromFile(filename string) (*x509.CertPool, error) {
+	certs, err := certificatesFromFile(filename)
+	if err != nil {
+		return nil, err
+	}
+	pool := x509.NewCertPool()
+	for _, cert := range certs {
+		pool.AddCert(cert)
+	}
+	return pool, nil
+}
+
+// certificatesFromFile returns the x509.Certificates contained in the given PEM-encoded file.
+// Returns an error if the file could not be read, a certificate could not be parsed, or if the file does not contain any certificates
+func certificatesFromFile(file string) ([]*x509.Certificate, error) {
+	if len(file) == 0 {
+		return nil, errors.New("error reading certificates from an empty filename")
+	}
+	pemBlock, err := ioutil.ReadFile(file)
+	if err != nil {
+		return nil, err
+	}
+	certs, err := CertsFromPEM(pemBlock)
+	if err != nil {
+		return nil, fmt.Errorf("error reading %s: %s", file, err)
+	}
+	return certs, nil
+}
+
+// CertsFromPEM returns the x509.Certificates contained in the given PEM-encoded byte array
+// Returns an error if a certificate could not be parsed, or if the data does not contain any certificates
+func CertsFromPEM(pemCerts []byte) ([]*x509.Certificate, error) {
+	ok := false
+	certs := []*x509.Certificate{}
+	for len(pemCerts) > 0 {
+		var block *pem.Block
+		block, pemCerts = pem.Decode(pemCerts)
+		if block == nil {
+			break
+		}
+		// Only use PEM "CERTIFICATE" blocks without extra headers
+		if block.Type != "CERTIFICATE" || len(block.Headers) != 0 {
+			continue
+		}
+
+		cert, err := x509.ParseCertificate(block.Bytes)
+		if err != nil {
+			return certs, err
+		}
+
+		certs = append(certs, cert)
+		ok = true
+	}
+
+	if !ok {
+		return certs, errors.New("could not read any certificates")
+	}
+	return certs, nil
+}

runtime/kubernetes/client/watch/body.go

@@ -0,0 +1,92 @@
+package watch
+
+import (
+	"bufio"
+	"encoding/json"
+	"net/http"
+	"time"
+)
+
+// bodyWatcher scans the body of a request for chunks
+type bodyWatcher struct {
+	results chan Event
+	stop    chan struct{}
+	res     *http.Response
+	req     *http.Request
+}
+
+// Changes returns the results channel
+func (wr *bodyWatcher) ResultChan() <-chan Event {
+	return wr.results
+}
+
+// Stop cancels the request
+func (wr *bodyWatcher) Stop() {
+	select {
+	case <-wr.stop:
+		return
+	default:
+		close(wr.stop)
+		close(wr.results)
+	}
+}
+
+func (wr *bodyWatcher) stream() {
+	reader := bufio.NewReader(wr.res.Body)
+
+	// ignore first few messages from stream,
+	// as they are usually old.
+	ignore := true
+
+	go func() {
+		<-time.After(time.Second)
+		ignore = false
+	}()
+
+	go func() {
+		// stop the watcher
+		defer wr.Stop()
+
+		for {
+			// read a line
+			b, err := reader.ReadBytes('\n')
+			if err != nil {
+				return
+			}
+
+			// ignore for the first second
+			if ignore {
+				continue
+			}
+
+			// send the event
+			var event Event
+			if err := json.Unmarshal(b, &event); err != nil {
+				continue
+			}
+			wr.results <- event
+		}
+	}()
+}
+
+// NewBodyWatcher creates a k8s body watcher for
+// a given http request
+func NewBodyWatcher(req *http.Request, client *http.Client) (Watch, error) {
+	stop := make(chan struct{})
+	req.Cancel = stop
+
+	res, err := client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+
+	wr := &bodyWatcher{
+		results: make(chan Event),
+		stop:    stop,
+		req:     req,
+		res:     res,
+	}
+
+	go wr.stream()
+	return wr, nil
+}

runtime/kubernetes/client/watch/watch.go

@@ -0,0 +1,26 @@
+package watch
+
+import "encoding/json"
+
+// Watch ...
+type Watch interface {
+	Stop()
+	ResultChan() <-chan Event
+}
+
+// EventType defines the possible types of events.
+type EventType string
+
+// EventTypes used
+const (
+	Added    EventType = "ADDED"
+	Modified EventType = "MODIFIED"
+	Deleted  EventType = "DELETED"
+	Error    EventType = "ERROR"
+)
+
+// Event represents a single event to a watched resource.
+type Event struct {
+	Type   EventType       `json:"type"`
+	Object json.RawMessage `json:"object"`
+}

runtime/kubernetes/client/watch/watch_test.go

@@ -0,0 +1,71 @@
+package watch
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+)
+
+var actions = []string{
+	`{"type": "create", "object":{"foo": "bar"}}`,
+	`{"type": "delete", INVALID}`,
+	`{"type": "update", "object":{"foo": {"foo": "bar"}}}`,
+	`{"type": "delete", "object":null}`,
+}
+
+func TestBodyWatcher(t *testing.T) {
+	// set up server with handler to flush strings from ch.
+	ch := make(chan string)
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		flusher, ok := w.(http.Flusher)
+		if !ok {
+			t.Fatal("expected ResponseWriter to be a flusher")
+		}
+
+		fmt.Fprintf(w, "\n")
+		flusher.Flush()
+
+		for v := range ch {
+			fmt.Fprintf(w, "%s\n", v)
+			flusher.Flush()
+			time.Sleep(10 * time.Millisecond)
+		}
+	}))
+	defer ts.Close()
+
+	req, err := http.NewRequest("GET", ts.URL, nil)
+	if err != nil {
+		t.Fatalf("did not expect NewRequest to return err: %v", err)
+	}
+
+	// setup body watcher
+	w, err := NewBodyWatcher(req, http.DefaultClient)
+	if err != nil {
+		t.Fatalf("did not expect NewBodyWatcher to return %v", err)
+	}
+
+	<-time.After(time.Second)
+
+	// send action strings in, and expect result back
+	ch <- actions[0]
+	if r := <-w.ResultChan(); r.Type != "create" {
+		t.Fatalf("expected result to be create")
+	}
+
+	ch <- actions[1] // should be ignored as its invalid json
+	ch <- actions[2]
+	if r := <-w.ResultChan(); r.Type != "update" {
+		t.Fatalf("expected result to be update")
+	}
+
+	ch <- actions[3]
+	if r := <-w.ResultChan(); r.Type != "delete" {
+		t.Fatalf("expected result to be delete")
+	}
+
+	// stop should clean up all channels.
+	w.Stop()
+	close(ch)
+}

runtime/kubernetes/kubernetes.go

@@ -0,0 +1,290 @@
+// Package kubernetes implements kubernetes micro runtime
+package kubernetes
+
+import (
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/micro/go-micro/runtime"
+	"github.com/micro/go-micro/runtime/kubernetes/client"
+	"github.com/micro/go-micro/util/log"
+)
+
+type kubernetes struct {
+	sync.RWMutex
+	// options configure runtime
+	options runtime.Options
+	// indicates if we're running
+	running bool
+	// used to start new services
+	start chan *runtime.Service
+	// used to stop the runtime
+	closed chan bool
+	// service tracks deployed services
+	services map[string]*runtime.Service
+	// client is kubernetes client
+	client client.Kubernetes
+}
+
+// NewRuntime creates new kubernetes runtime
+func NewRuntime(opts ...runtime.Option) runtime.Runtime {
+	// get default options
+	options := runtime.Options{}
+
+	// apply requested options
+	for _, o := range opts {
+		o(&options)
+	}
+
+	// kubernetes client
+	client := client.NewClientInCluster()
+
+	return &kubernetes{
+		options:  options,
+		closed:   make(chan bool),
+		start:    make(chan *runtime.Service, 128),
+		services: make(map[string]*runtime.Service),
+		client:   client,
+	}
+}
+
+// Init initializes runtime options
+func (k *kubernetes) Init(opts ...runtime.Option) error {
+	k.Lock()
+	defer k.Unlock()
+
+	for _, o := range opts {
+		o(&k.options)
+	}
+
+	return nil
+}
+
+// Registers a service
+func (k *kubernetes) Create(s *runtime.Service, opts ...runtime.CreateOption) error {
+	k.Lock()
+	defer k.Unlock()
+
+	// TODO:
+	// * create service
+	// * create deployment
+
+	// NOTE: our services have micro- prefix
+	muName := strings.Split(s.Name, ".")
+	s.Name = "micro-" + muName[len(muName)-1]
+
+	// NOTE: we are tracking this in memory for now
+	if _, ok := k.services[s.Name]; ok {
+		return errors.New("service already registered")
+	}
+
+	var options runtime.CreateOptions
+	for _, o := range opts {
+		o(&options)
+	}
+
+	// save service
+	k.services[s.Name] = s
+	// push into start queue
+	k.start <- k.services[s.Name]
+
+	return nil
+}
+
+// Remove a service
+func (k *kubernetes) Delete(s *runtime.Service) error {
+	k.Lock()
+	defer k.Unlock()
+
+	// TODO:
+	// * delete service
+	// * delete dpeloyment
+
+	// NOTE: we are tracking this in memory for now
+	if s, ok := k.services[s.Name]; ok {
+		delete(k.services, s.Name)
+		return nil
+	}
+
+	return nil
+}
+
+// Update the service in place
+func (k *kubernetes) Update(s *runtime.Service) error {
+	type body struct {
+		Metadata *client.Metadata `json:"metadata"`
+	}
+	// parse version into human readable timestamp
+	updateTimeStamp, err := strconv.ParseInt(s.Version, 10, 64)
+	if err != nil {
+		return err
+	}
+	unixTimeUTC := time.Unix(updateTimeStamp, 0)
+	// metada which we will PATCH deployment with
+	reqBody := body{
+		Metadata: &client.Metadata{
+			Annotations: map[string]string{
+				"build": unixTimeUTC.Format(time.RFC3339),
+			},
+		},
+	}
+	return k.client.UpdateDeployment(s.Name, reqBody)
+}
+
+// List the managed services
+func (k *kubernetes) List() ([]*runtime.Service, error) {
+	// TODO: this should list the k8s deployments
+	// but for now we return in-memory tracked services
+	var services []*runtime.Service
+	k.RLock()
+	defer k.RUnlock()
+
+	for _, service := range k.services {
+		services = append(services, service)
+	}
+
+	return services, nil
+}
+
+// run runs the runtime management loop
+func (k *kubernetes) run(events <-chan runtime.Event) {
+	t := time.NewTicker(time.Second * 5)
+	defer t.Stop()
+
+	for {
+		select {
+		case <-t.C:
+			// TODO: noop for now
+			// check running services
+			// * deployments exist
+			// * service is exposed
+		case service := <-k.start:
+			// TODO: following might have to be done
+			// * create a deployment
+			// * expose a service
+			log.Debugf("Runtime starting service: %s", service.Name)
+		case event := <-events:
+			// NOTE: we only handle Update events for now
+			log.Debugf("Runtime received notification event: %v", event)
+			switch event.Type {
+			case runtime.Update:
+				// parse returned response to timestamp
+				updateTimeStamp, err := strconv.ParseInt(event.Version, 10, 64)
+				if err != nil {
+					log.Debugf("Runtime error parsing update build time: %v", err)
+					continue
+				}
+				buildTime := time.Unix(updateTimeStamp, 0)
+				processEvent := func(event runtime.Event, service *runtime.Service) error {
+					buildTimeStamp, err := strconv.ParseInt(service.Version, 10, 64)
+					if err != nil {
+						return err
+					}
+					muBuild := time.Unix(buildTimeStamp, 0)
+					if buildTime.After(muBuild) {
+						version := fmt.Sprintf("%d", buildTime.Unix())
+						muService := &runtime.Service{
+							Name:    service.Name,
+							Source:  service.Source,
+							Path:    service.Path,
+							Exec:    service.Exec,
+							Version: version,
+						}
+						if err := k.Update(muService); err != nil {
+							return err
+						}
+						service.Version = version
+					}
+					return nil
+				}
+				k.Lock()
+				if len(event.Service) > 0 {
+					service, ok := k.services[event.Service]
+					if !ok {
+						log.Debugf("Runtime unknown service: %s", event.Service)
+						k.Unlock()
+						continue
+					}
+					if err := processEvent(event, service); err != nil {
+						log.Debugf("Runtime error updating service %s: %v", event.Service, err)
+					}
+					k.Unlock()
+					continue
+				}
+				// if blank service was received we update all services
+				for _, service := range k.services {
+					if err := processEvent(event, service); err != nil {
+						log.Debugf("Runtime error updating service %s: %v", service.Name, err)
+					}
+				}
+				k.Unlock()
+			}
+		case <-k.closed:
+			log.Debugf("Runtime stopped")
+			return
+		}
+	}
+}
+
+// starts the runtime
+func (k *kubernetes) Start() error {
+	k.Lock()
+	defer k.Unlock()
+
+	// already running
+	if k.running {
+		return nil
+	}
+
+	// set running
+	k.running = true
+	k.closed = make(chan bool)
+
+	var events <-chan runtime.Event
+	if k.options.Notifier != nil {
+		var err error
+		events, err = k.options.Notifier.Notify()
+		if err != nil {
+			// TODO: should we bail here?
+			log.Debugf("Runtime failed to start update notifier")
+		}
+	}
+
+	go k.run(events)
+
+	return nil
+}
+
+// Shutdown the runtime
+func (k *kubernetes) Stop() error {
+	k.Lock()
+	defer k.Unlock()
+
+	if !k.running {
+		return nil
+	}
+
+	select {
+	case <-k.closed:
+		return nil
+	default:
+		close(k.closed)
+		// set not running
+		k.running = false
+		// stop the notifier too
+		if k.options.Notifier != nil {
+			return k.options.Notifier.Close()
+		}
+	}
+
+	return nil
+}
+
+// String implements stringer interface
+func (k *kubernetes) String() string {
+	return "kubernetes"
+}

runtime/options.go

@@ -4,8 +4,24 @@ import (
 	"io"
 )
 
+type Option func(o *Options)
+
+// Options configure runtime
+type Options struct {
+	// Notifier for updates
+	Notifier Notifier
+}
+
+// AutoUpdate enables micro auto-updates
+func WithNotifier(n Notifier) Option {
+	return func(o *Options) {
+		o.Notifier = n
+	}
+}
+
 type CreateOption func(o *CreateOptions)
 
+// CreateOptions configure runtime services
 type CreateOptions struct {
 	// command to execute including args
 	Command []string
@@ -25,7 +41,7 @@ func WithCommand(c string, args ...string) CreateOption {
 	}
 }
 
-// WithEnv sets the created service env
+// WithEnv sets the created service environment
 func WithEnv(env []string) CreateOption {
 	return func(o *CreateOptions) {
 		o.Env = env

runtime/runtime.go

@@ -1,45 +1,87 @@
 // Package runtime is a service runtime manager
 package runtime
 
+import "time"
+
+var (
+	// DefaultRuntime is default micro runtime
+	DefaultRuntime Runtime = NewRuntime()
+)
+
 // Runtime is a service runtime manager
 type Runtime interface {
+	// Init initializes runtime
+	Init(...Option) error
 	// Registers a service
 	Create(*Service, ...CreateOption) error
 	// Remove a service
 	Delete(*Service) error
+	// Update the service in place
+	Update(*Service) error
+	// List the managed services
+	List() ([]*Service, error)
 	// starts the runtime
 	Start() error
 	// Shutdown the runtime
 	Stop() error
 }
 
+// Notifier is an update notifier
+type Notifier interface {
+	// Notify publishes notification events
+	Notify() (<-chan Event, error)
+	// Close stops the notifier
+	Close() error
+}
+
+// EventType defines notification event
+type EventType int
+
+const (
+	// Create is emitted when a new build has been craeted
+	Create EventType = iota
+	// Update is emitted when a new update become available
+	Update
+	// Delete is emitted when a build has been deleted
+	Delete
+)
+
+// String returns human readable event type
+func (t EventType) String() string {
+	switch t {
+	case Create:
+		return "create"
+	case Delete:
+		return "delete"
+	case Update:
+		return "update"
+	default:
+		return "unknown"
+	}
+}
+
+// Event is notification event
+type Event struct {
+	// Type is event type
+	Type EventType
+	// Timestamp is event timestamp
+	Timestamp time.Time
+	// Service is the name of the service
+	Service string
+	// Version of the build
+	Version string
+}
+
+// Service is runtime service
 type Service struct {
-	// name of the service
+	// Name of the service
 	Name string
 	// url location of source
 	Source string
-	// path to store source
+	// Path to store source
 	Path string
-	// exec command
+	// Exec command
 	Exec string
-}
-
-var (
-	DefaultRuntime = newRuntime()
-)
-
-func Create(s *Service, opts ...CreateOption) error {
-	return DefaultRuntime.Create(s, opts...)
-}
-
-func Delete(s *Service) error {
-	return DefaultRuntime.Delete(s)
-}
-
-func Start() error {
-	return DefaultRuntime.Start()
-}
-
-func Stop() error {
-	return DefaultRuntime.Stop()
+	// Version of the service
+	Version string
 }

runtime/service.go

@@ -0,0 +1,158 @@
+package runtime
+
+import (
+	"io"
+	"strings"
+	"sync"
+
+	packager "github.com/micro/go-micro/runtime/package"
+	"github.com/micro/go-micro/runtime/process"
+	proc "github.com/micro/go-micro/runtime/process/os"
+	"github.com/micro/go-micro/util/log"
+)
+
+type service struct {
+	sync.RWMutex
+
+	running bool
+	closed  chan bool
+	err     error
+
+	// output for logs
+	output io.Writer
+
+	// service to manage
+	*Service
+	// process creator
+	Process *proc.Process
+	// Exec
+	Exec *process.Executable
+	// process pid
+	PID *process.PID
+}
+
+func newService(s *Service, c CreateOptions) *service {
+	var exec string
+	var args []string
+
+	if len(s.Exec) > 0 {
+		parts := strings.Split(s.Exec, " ")
+		exec = parts[0]
+		args = []string{}
+
+		if len(parts) > 1 {
+			args = parts[1:]
+		}
+	} else {
+		// set command
+		exec = c.Command[0]
+		// set args
+		if len(c.Command) > 1 {
+			args = c.Command[1:]
+		}
+	}
+
+	return &service{
+		Service: s,
+		Process: new(proc.Process),
+		Exec: &process.Executable{
+			Binary: &packager.Binary{
+				Name: s.Name,
+				Path: exec,
+			},
+			Env:  c.Env,
+			Args: args,
+		},
+		closed: make(chan bool),
+		output: c.Output,
+	}
+}
+
+func (s *service) streamOutput() {
+	go io.Copy(s.output, s.PID.Output)
+	go io.Copy(s.output, s.PID.Error)
+}
+
+// Running returns true is the service is running
+func (s *service) Running() bool {
+	s.RLock()
+	defer s.RUnlock()
+	return s.running
+}
+
+// Start stars the service
+func (s *service) Start() error {
+	s.Lock()
+	defer s.Unlock()
+
+	if s.running {
+		return nil
+	}
+
+	// reset
+	s.err = nil
+	s.closed = make(chan bool)
+
+	// TODO: pull source & build binary
+	log.Debugf("Runtime service %s forking new process", s.Service.Name)
+	p, err := s.Process.Fork(s.Exec)
+	if err != nil {
+		return err
+	}
+
+	// set the pid
+	s.PID = p
+	// set to running
+	s.running = true
+
+	if s.output != nil {
+		s.streamOutput()
+	}
+
+	// wait and watch
+	go s.Wait()
+
+	return nil
+}
+
+// Stop stops the service
+func (s *service) Stop() error {
+	s.Lock()
+	defer s.Unlock()
+
+	select {
+	case <-s.closed:
+		return nil
+	default:
+		close(s.closed)
+		s.running = false
+		if s.PID == nil {
+			return nil
+		}
+		return s.Process.Kill(s.PID)
+	}
+}
+
+// Error returns the last error service has returned
+func (s *service) Error() error {
+	s.RLock()
+	defer s.RUnlock()
+	return s.err
+}
+
+// Wait waits for the service to finish running
+func (s *service) Wait() {
+	// wait for process to exit
+	err := s.Process.Wait(s.PID)
+
+	s.Lock()
+	defer s.Unlock()
+
+	// save the error
+	if err != nil {
+		s.err = err
+	}
+
+	// no longer running
+	s.running = false
+}

runtime/service/handler/handler.go

@@ -0,0 +1,91 @@
+package handler
+
+import (
+	"context"
+
+	"github.com/micro/go-micro/errors"
+	"github.com/micro/go-micro/runtime"
+	pb "github.com/micro/go-micro/runtime/service/proto"
+)
+
+type Runtime struct {
+	Runtime runtime.Runtime
+}
+
+func toProto(s *runtime.Service) *pb.Service {
+	return &pb.Service{
+		Name:    s.Name,
+		Version: s.Version,
+		Source:  s.Source,
+		Path:    s.Path,
+		Exec:    s.Exec,
+	}
+}
+
+func toService(s *pb.Service) *runtime.Service {
+	return &runtime.Service{
+		Name:    s.Name,
+		Version: s.Version,
+		Source:  s.Source,
+		Path:    s.Path,
+		Exec:    s.Exec,
+	}
+}
+
+func (r *Runtime) Create(ctx context.Context, req *pb.CreateRequest, rsp *pb.CreateResponse) error {
+	if req.Service == nil {
+		return errors.BadRequest("go.micro.runtime", "blank service")
+	}
+
+	// TODO: add opts
+	service := toService(req.Service)
+	err := r.Runtime.Create(service)
+	if err != nil {
+		return errors.InternalServerError("go.micro.runtime", err.Error())
+	}
+
+	return nil
+}
+
+func (r *Runtime) Update(ctx context.Context, req *pb.UpdateRequest, rsp *pb.UpdateResponse) error {
+	if req.Service == nil {
+		return errors.BadRequest("go.micro.runtime", "blank service")
+	}
+
+	// TODO: add opts
+	service := toService(req.Service)
+	err := r.Runtime.Update(service)
+	if err != nil {
+		return errors.InternalServerError("go.micro.runtime", err.Error())
+	}
+
+	return nil
+}
+
+func (r *Runtime) Delete(ctx context.Context, req *pb.DeleteRequest, rsp *pb.DeleteResponse) error {
+	if req.Service == nil {
+		return errors.BadRequest("go.micro.runtime", "blank service")
+	}
+
+	// TODO: add opts
+	service := toService(req.Service)
+	err := r.Runtime.Delete(service)
+	if err != nil {
+		return errors.InternalServerError("go.micro.runtime", err.Error())
+	}
+
+	return nil
+}
+
+func (r *Runtime) List(ctx context.Context, req *pb.ListRequest, rsp *pb.ListResponse) error {
+	services, err := r.Runtime.List()
+	if err != nil {
+		return errors.InternalServerError("go.micro.runtime", err.Error())
+	}
+
+	for _, service := range services {
+		rsp.Services = append(rsp.Services, toProto(service))
+	}
+
+	return nil
+}

runtime/service/proto/runtime.micro.go

@@ -0,0 +1,142 @@
+// Code generated by protoc-gen-micro. DO NOT EDIT.
+// source: micro/go-micro/runtime/service/proto/runtime.proto
+
+package go_micro_runtime
+
+import (
+	fmt "fmt"
+	proto "github.com/golang/protobuf/proto"
+	math "math"
+)
+
+import (
+	context "context"
+	client "github.com/micro/go-micro/client"
+	server "github.com/micro/go-micro/server"
+)
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ = proto.Marshal
+var _ = fmt.Errorf
+var _ = math.Inf
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the proto package it is being compiled against.
+// A compilation error at this line likely means your copy of the
+// proto package needs to be updated.
+const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ context.Context
+var _ client.Option
+var _ server.Option
+
+// Client API for Runtime service
+
+type RuntimeService interface {
+	Create(ctx context.Context, in *CreateRequest, opts ...client.CallOption) (*CreateResponse, error)
+	Delete(ctx context.Context, in *DeleteRequest, opts ...client.CallOption) (*DeleteResponse, error)
+	Update(ctx context.Context, in *UpdateRequest, opts ...client.CallOption) (*UpdateResponse, error)
+	List(ctx context.Context, in *ListRequest, opts ...client.CallOption) (*ListResponse, error)
+}
+
+type runtimeService struct {
+	c    client.Client
+	name string
+}
+
+func NewRuntimeService(name string, c client.Client) RuntimeService {
+	if c == nil {
+		c = client.NewClient()
+	}
+	if len(name) == 0 {
+		name = "go.micro.runtime"
+	}
+	return &runtimeService{
+		c:    c,
+		name: name,
+	}
+}
+
+func (c *runtimeService) Create(ctx context.Context, in *CreateRequest, opts ...client.CallOption) (*CreateResponse, error) {
+	req := c.c.NewRequest(c.name, "Runtime.Create", in)
+	out := new(CreateResponse)
+	err := c.c.Call(ctx, req, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *runtimeService) Delete(ctx context.Context, in *DeleteRequest, opts ...client.CallOption) (*DeleteResponse, error) {
+	req := c.c.NewRequest(c.name, "Runtime.Delete", in)
+	out := new(DeleteResponse)
+	err := c.c.Call(ctx, req, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *runtimeService) Update(ctx context.Context, in *UpdateRequest, opts ...client.CallOption) (*UpdateResponse, error) {
+	req := c.c.NewRequest(c.name, "Runtime.Update", in)
+	out := new(UpdateResponse)
+	err := c.c.Call(ctx, req, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *runtimeService) List(ctx context.Context, in *ListRequest, opts ...client.CallOption) (*ListResponse, error) {
+	req := c.c.NewRequest(c.name, "Runtime.List", in)
+	out := new(ListResponse)
+	err := c.c.Call(ctx, req, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// Server API for Runtime service
+
+type RuntimeHandler interface {
+	Create(context.Context, *CreateRequest, *CreateResponse) error
+	Delete(context.Context, *DeleteRequest, *DeleteResponse) error
+	Update(context.Context, *UpdateRequest, *UpdateResponse) error
+	List(context.Context, *ListRequest, *ListResponse) error
+}
+
+func RegisterRuntimeHandler(s server.Server, hdlr RuntimeHandler, opts ...server.HandlerOption) error {
+	type runtime interface {
+		Create(ctx context.Context, in *CreateRequest, out *CreateResponse) error
+		Delete(ctx context.Context, in *DeleteRequest, out *DeleteResponse) error
+		Update(ctx context.Context, in *UpdateRequest, out *UpdateResponse) error
+		List(ctx context.Context, in *ListRequest, out *ListResponse) error
+	}
+	type Runtime struct {
+		runtime
+	}
+	h := &runtimeHandler{hdlr}
+	return s.Handle(s.NewHandler(&Runtime{h}, opts...))
+}
+
+type runtimeHandler struct {
+	RuntimeHandler
+}
+
+func (h *runtimeHandler) Create(ctx context.Context, in *CreateRequest, out *CreateResponse) error {
+	return h.RuntimeHandler.Create(ctx, in, out)
+}
+
+func (h *runtimeHandler) Delete(ctx context.Context, in *DeleteRequest, out *DeleteResponse) error {
+	return h.RuntimeHandler.Delete(ctx, in, out)
+}
+
+func (h *runtimeHandler) Update(ctx context.Context, in *UpdateRequest, out *UpdateResponse) error {
+	return h.RuntimeHandler.Update(ctx, in, out)
+}
+
+func (h *runtimeHandler) List(ctx context.Context, in *ListRequest, out *ListResponse) error {
+	return h.RuntimeHandler.List(ctx, in, out)
+}

runtime/service/proto/runtime.pb.go

@@ -0,0 +1,661 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// source: micro/go-micro/runtime/service/proto/runtime.proto
+
+package go_micro_runtime
+
+import (
+	context "context"
+	fmt "fmt"
+	proto "github.com/golang/protobuf/proto"
+	grpc "google.golang.org/grpc"
+	math "math"
+)
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ = proto.Marshal
+var _ = fmt.Errorf
+var _ = math.Inf
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the proto package it is being compiled against.
+// A compilation error at this line likely means your copy of the
+// proto package needs to be updated.
+const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package
+
+type Service struct {
+	// name of the service
+	Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
+	// version of the service
+	Version string `protobuf:"bytes,2,opt,name=version,proto3" json:"version,omitempty"`
+	// git url of the source
+	Source string `protobuf:"bytes,3,opt,name=source,proto3" json:"source,omitempty"`
+	// local path of the source
+	Path string `protobuf:"bytes,4,opt,name=path,proto3" json:"path,omitempty"`
+	// command to execute
+	Exec                 string   `protobuf:"bytes,5,opt,name=exec,proto3" json:"exec,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *Service) Reset()         { *m = Service{} }
+func (m *Service) String() string { return proto.CompactTextString(m) }
+func (*Service) ProtoMessage()    {}
+func (*Service) Descriptor() ([]byte, []int) {
+	return fileDescriptor_4bc91a8efec81434, []int{0}
+}
+
+func (m *Service) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_Service.Unmarshal(m, b)
+}
+func (m *Service) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_Service.Marshal(b, m, deterministic)
+}
+func (m *Service) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_Service.Merge(m, src)
+}
+func (m *Service) XXX_Size() int {
+	return xxx_messageInfo_Service.Size(m)
+}
+func (m *Service) XXX_DiscardUnknown() {
+	xxx_messageInfo_Service.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_Service proto.InternalMessageInfo
+
+func (m *Service) GetName() string {
+	if m != nil {
+		return m.Name
+	}
+	return ""
+}
+
+func (m *Service) GetVersion() string {
+	if m != nil {
+		return m.Version
+	}
+	return ""
+}
+
+func (m *Service) GetSource() string {
+	if m != nil {
+		return m.Source
+	}
+	return ""
+}
+
+func (m *Service) GetPath() string {
+	if m != nil {
+		return m.Path
+	}
+	return ""
+}
+
+func (m *Service) GetExec() string {
+	if m != nil {
+		return m.Exec
+	}
+	return ""
+}
+
+type Options struct {
+	// command to pass in
+	Command string `protobuf:"bytes,1,opt,name=command,proto3" json:"command,omitempty"`
+	// environment to pass in
+	Env []string `protobuf:"bytes,2,rep,name=env,proto3" json:"env,omitempty"`
+	// output to send to
+	Output               string   `protobuf:"bytes,3,opt,name=output,proto3" json:"output,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *Options) Reset()         { *m = Options{} }
+func (m *Options) String() string { return proto.CompactTextString(m) }
+func (*Options) ProtoMessage()    {}
+func (*Options) Descriptor() ([]byte, []int) {
+	return fileDescriptor_4bc91a8efec81434, []int{1}
+}
+
+func (m *Options) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_Options.Unmarshal(m, b)
+}
+func (m *Options) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_Options.Marshal(b, m, deterministic)
+}
+func (m *Options) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_Options.Merge(m, src)
+}
+func (m *Options) XXX_Size() int {
+	return xxx_messageInfo_Options.Size(m)
+}
+func (m *Options) XXX_DiscardUnknown() {
+	xxx_messageInfo_Options.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_Options proto.InternalMessageInfo
+
+func (m *Options) GetCommand() string {
+	if m != nil {
+		return m.Command
+	}
+	return ""
+}
+
+func (m *Options) GetEnv() []string {
+	if m != nil {
+		return m.Env
+	}
+	return nil
+}
+
+func (m *Options) GetOutput() string {
+	if m != nil {
+		return m.Output
+	}
+	return ""
+}
+
+type CreateRequest struct {
+	Service              *Service `protobuf:"bytes,1,opt,name=service,proto3" json:"service,omitempty"`
+	Options              *Options `protobuf:"bytes,2,opt,name=options,proto3" json:"options,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *CreateRequest) Reset()         { *m = CreateRequest{} }
+func (m *CreateRequest) String() string { return proto.CompactTextString(m) }
+func (*CreateRequest) ProtoMessage()    {}
+func (*CreateRequest) Descriptor() ([]byte, []int) {
+	return fileDescriptor_4bc91a8efec81434, []int{2}
+}
+
+func (m *CreateRequest) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_CreateRequest.Unmarshal(m, b)
+}
+func (m *CreateRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_CreateRequest.Marshal(b, m, deterministic)
+}
+func (m *CreateRequest) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_CreateRequest.Merge(m, src)
+}
+func (m *CreateRequest) XXX_Size() int {
+	return xxx_messageInfo_CreateRequest.Size(m)
+}
+func (m *CreateRequest) XXX_DiscardUnknown() {
+	xxx_messageInfo_CreateRequest.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_CreateRequest proto.InternalMessageInfo
+
+func (m *CreateRequest) GetService() *Service {
+	if m != nil {
+		return m.Service
+	}
+	return nil
+}
+
+func (m *CreateRequest) GetOptions() *Options {
+	if m != nil {
+		return m.Options
+	}
+	return nil
+}
+
+type CreateResponse struct {
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *CreateResponse) Reset()         { *m = CreateResponse{} }
+func (m *CreateResponse) String() string { return proto.CompactTextString(m) }
+func (*CreateResponse) ProtoMessage()    {}
+func (*CreateResponse) Descriptor() ([]byte, []int) {
+	return fileDescriptor_4bc91a8efec81434, []int{3}
+}
+
+func (m *CreateResponse) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_CreateResponse.Unmarshal(m, b)
+}
+func (m *CreateResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_CreateResponse.Marshal(b, m, deterministic)
+}
+func (m *CreateResponse) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_CreateResponse.Merge(m, src)
+}
+func (m *CreateResponse) XXX_Size() int {
+	return xxx_messageInfo_CreateResponse.Size(m)
+}
+func (m *CreateResponse) XXX_DiscardUnknown() {
+	xxx_messageInfo_CreateResponse.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_CreateResponse proto.InternalMessageInfo
+
+type DeleteRequest struct {
+	Service              *Service `protobuf:"bytes,1,opt,name=service,proto3" json:"service,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *DeleteRequest) Reset()         { *m = DeleteRequest{} }
+func (m *DeleteRequest) String() string { return proto.CompactTextString(m) }
+func (*DeleteRequest) ProtoMessage()    {}
+func (*DeleteRequest) Descriptor() ([]byte, []int) {
+	return fileDescriptor_4bc91a8efec81434, []int{4}
+}
+
+func (m *DeleteRequest) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_DeleteRequest.Unmarshal(m, b)
+}
+func (m *DeleteRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_DeleteRequest.Marshal(b, m, deterministic)
+}
+func (m *DeleteRequest) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_DeleteRequest.Merge(m, src)
+}
+func (m *DeleteRequest) XXX_Size() int {
+	return xxx_messageInfo_DeleteRequest.Size(m)
+}
+func (m *DeleteRequest) XXX_DiscardUnknown() {
+	xxx_messageInfo_DeleteRequest.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_DeleteRequest proto.InternalMessageInfo
+
+func (m *DeleteRequest) GetService() *Service {
+	if m != nil {
+		return m.Service
+	}
+	return nil
+}
+
+type DeleteResponse struct {
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *DeleteResponse) Reset()         { *m = DeleteResponse{} }
+func (m *DeleteResponse) String() string { return proto.CompactTextString(m) }
+func (*DeleteResponse) ProtoMessage()    {}
+func (*DeleteResponse) Descriptor() ([]byte, []int) {
+	return fileDescriptor_4bc91a8efec81434, []int{5}
+}
+
+func (m *DeleteResponse) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_DeleteResponse.Unmarshal(m, b)
+}
+func (m *DeleteResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_DeleteResponse.Marshal(b, m, deterministic)
+}
+func (m *DeleteResponse) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_DeleteResponse.Merge(m, src)
+}
+func (m *DeleteResponse) XXX_Size() int {
+	return xxx_messageInfo_DeleteResponse.Size(m)
+}
+func (m *DeleteResponse) XXX_DiscardUnknown() {
+	xxx_messageInfo_DeleteResponse.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_DeleteResponse proto.InternalMessageInfo
+
+type UpdateRequest struct {
+	Service              *Service `protobuf:"bytes,1,opt,name=service,proto3" json:"service,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *UpdateRequest) Reset()         { *m = UpdateRequest{} }
+func (m *UpdateRequest) String() string { return proto.CompactTextString(m) }
+func (*UpdateRequest) ProtoMessage()    {}
+func (*UpdateRequest) Descriptor() ([]byte, []int) {
+	return fileDescriptor_4bc91a8efec81434, []int{6}
+}
+
+func (m *UpdateRequest) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_UpdateRequest.Unmarshal(m, b)
+}
+func (m *UpdateRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_UpdateRequest.Marshal(b, m, deterministic)
+}
+func (m *UpdateRequest) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_UpdateRequest.Merge(m, src)
+}
+func (m *UpdateRequest) XXX_Size() int {
+	return xxx_messageInfo_UpdateRequest.Size(m)
+}
+func (m *UpdateRequest) XXX_DiscardUnknown() {
+	xxx_messageInfo_UpdateRequest.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_UpdateRequest proto.InternalMessageInfo
+
+func (m *UpdateRequest) GetService() *Service {
+	if m != nil {
+		return m.Service
+	}
+	return nil
+}
+
+type UpdateResponse struct {
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *UpdateResponse) Reset()         { *m = UpdateResponse{} }
+func (m *UpdateResponse) String() string { return proto.CompactTextString(m) }
+func (*UpdateResponse) ProtoMessage()    {}
+func (*UpdateResponse) Descriptor() ([]byte, []int) {
+	return fileDescriptor_4bc91a8efec81434, []int{7}
+}
+
+func (m *UpdateResponse) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_UpdateResponse.Unmarshal(m, b)
+}
+func (m *UpdateResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_UpdateResponse.Marshal(b, m, deterministic)
+}
+func (m *UpdateResponse) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_UpdateResponse.Merge(m, src)
+}
+func (m *UpdateResponse) XXX_Size() int {
+	return xxx_messageInfo_UpdateResponse.Size(m)
+}
+func (m *UpdateResponse) XXX_DiscardUnknown() {
+	xxx_messageInfo_UpdateResponse.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_UpdateResponse proto.InternalMessageInfo
+
+type ListRequest struct {
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *ListRequest) Reset()         { *m = ListRequest{} }
+func (m *ListRequest) String() string { return proto.CompactTextString(m) }
+func (*ListRequest) ProtoMessage()    {}
+func (*ListRequest) Descriptor() ([]byte, []int) {
+	return fileDescriptor_4bc91a8efec81434, []int{8}
+}
+
+func (m *ListRequest) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_ListRequest.Unmarshal(m, b)
+}
+func (m *ListRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_ListRequest.Marshal(b, m, deterministic)
+}
+func (m *ListRequest) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_ListRequest.Merge(m, src)
+}
+func (m *ListRequest) XXX_Size() int {
+	return xxx_messageInfo_ListRequest.Size(m)
+}
+func (m *ListRequest) XXX_DiscardUnknown() {
+	xxx_messageInfo_ListRequest.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_ListRequest proto.InternalMessageInfo
+
+type ListResponse struct {
+	Services             []*Service `protobuf:"bytes,1,rep,name=services,proto3" json:"services,omitempty"`
+	XXX_NoUnkeyedLiteral struct{}   `json:"-"`
+	XXX_unrecognized     []byte     `json:"-"`
+	XXX_sizecache        int32      `json:"-"`
+}
+
+func (m *ListResponse) Reset()         { *m = ListResponse{} }
+func (m *ListResponse) String() string { return proto.CompactTextString(m) }
+func (*ListResponse) ProtoMessage()    {}
+func (*ListResponse) Descriptor() ([]byte, []int) {
+	return fileDescriptor_4bc91a8efec81434, []int{9}
+}
+
+func (m *ListResponse) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_ListResponse.Unmarshal(m, b)
+}
+func (m *ListResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_ListResponse.Marshal(b, m, deterministic)
+}
+func (m *ListResponse) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_ListResponse.Merge(m, src)
+}
+func (m *ListResponse) XXX_Size() int {
+	return xxx_messageInfo_ListResponse.Size(m)
+}
+func (m *ListResponse) XXX_DiscardUnknown() {
+	xxx_messageInfo_ListResponse.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_ListResponse proto.InternalMessageInfo
+
+func (m *ListResponse) GetServices() []*Service {
+	if m != nil {
+		return m.Services
+	}
+	return nil
+}
+
+func init() {
+	proto.RegisterType((*Service)(nil), "go.micro.runtime.Service")
+	proto.RegisterType((*Options)(nil), "go.micro.runtime.Options")
+	proto.RegisterType((*CreateRequest)(nil), "go.micro.runtime.CreateRequest")
+	proto.RegisterType((*CreateResponse)(nil), "go.micro.runtime.CreateResponse")
+	proto.RegisterType((*DeleteRequest)(nil), "go.micro.runtime.DeleteRequest")
+	proto.RegisterType((*DeleteResponse)(nil), "go.micro.runtime.DeleteResponse")
+	proto.RegisterType((*UpdateRequest)(nil), "go.micro.runtime.UpdateRequest")
+	proto.RegisterType((*UpdateResponse)(nil), "go.micro.runtime.UpdateResponse")
+	proto.RegisterType((*ListRequest)(nil), "go.micro.runtime.ListRequest")
+	proto.RegisterType((*ListResponse)(nil), "go.micro.runtime.ListResponse")
+}
+
+func init() {
+	proto.RegisterFile("micro/go-micro/runtime/service/proto/runtime.proto", fileDescriptor_4bc91a8efec81434)
+}
+
+var fileDescriptor_4bc91a8efec81434 = []byte{
+	// 379 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xa4, 0x93, 0xcb, 0x4e, 0xc2, 0x40,
+	0x14, 0x86, 0x81, 0x22, 0x95, 0x83, 0x18, 0x32, 0x0b, 0x33, 0x92, 0xa8, 0x4d, 0x57, 0x6c, 0x2c,
+	0x09, 0xc4, 0x27, 0x10, 0xe3, 0x46, 0x62, 0x52, 0xe3, 0x03, 0xd4, 0x72, 0x82, 0x4d, 0x6c, 0xa7,
+	0xce, 0x4c, 0x51, 0x9f, 0xce, 0x57, 0x33, 0x73, 0x43, 0x81, 0xe2, 0x86, 0xdd, 0xb9, 0x7e, 0xff,
+	0x99, 0xbf, 0x29, 0x4c, 0xf2, 0x2c, 0xe5, 0x6c, 0xbc, 0x64, 0xd7, 0x26, 0xe0, 0x55, 0x21, 0xb3,
+	0x1c, 0xc7, 0x02, 0xf9, 0x2a, 0x4b, 0x71, 0x5c, 0x72, 0x26, 0xd7, 0xd5, 0x48, 0x67, 0x64, 0xb0,
+	0x64, 0x91, 0x9e, 0x8e, 0x6c, 0x3d, 0xfc, 0x00, 0xff, 0xc9, 0x2c, 0x10, 0x02, 0xed, 0x22, 0xc9,
+	0x91, 0x36, 0x83, 0xe6, 0xa8, 0x1b, 0xeb, 0x98, 0x50, 0xf0, 0x57, 0xc8, 0x45, 0xc6, 0x0a, 0xda,
+	0xd2, 0x65, 0x97, 0x92, 0x33, 0xe8, 0x08, 0x56, 0xf1, 0x14, 0xa9, 0xa7, 0x1b, 0x36, 0x53, 0x94,
+	0x32, 0x91, 0xaf, 0xb4, 0x6d, 0x28, 0x2a, 0x56, 0x35, 0xfc, 0xc4, 0x94, 0x1e, 0x99, 0x9a, 0x8a,
+	0xc3, 0x39, 0xf8, 0x8f, 0xa5, 0xcc, 0x58, 0x21, 0x94, 0x48, 0xca, 0xf2, 0x3c, 0x29, 0x16, 0x56,
+	0xdb, 0xa5, 0x64, 0x00, 0x1e, 0x16, 0x2b, 0xda, 0x0a, 0xbc, 0x51, 0x37, 0x56, 0xa1, 0x92, 0x65,
+	0x95, 0x2c, 0x2b, 0xe9, 0x64, 0x4d, 0x16, 0x7e, 0x41, 0xff, 0x96, 0x63, 0x22, 0x31, 0xc6, 0xf7,
+	0x0a, 0x85, 0x24, 0x53, 0xf0, 0xad, 0x13, 0x1a, 0xda, 0x9b, 0x9c, 0x47, 0xdb, 0x8f, 0x8f, 0xec,
+	0xcb, 0x63, 0x37, 0xa9, 0x96, 0x98, 0x39, 0x4a, 0x3f, 0xb7, 0x76, 0xc9, 0x5e, 0x1d, 0xbb, 0xc9,
+	0x70, 0x00, 0xa7, 0x4e, 0x5a, 0x94, 0xac, 0x10, 0x18, 0xce, 0xa0, 0x3f, 0xc3, 0x37, 0x3c, 0xec,
+	0x18, 0xc5, 0x75, 0x94, 0x5f, 0xee, 0x73, 0xb9, 0x48, 0x0e, 0xe7, 0x3a, 0x8a, 0xe5, 0xf6, 0xa1,
+	0xf7, 0x90, 0x09, 0x69, 0xa9, 0xe1, 0x1d, 0x9c, 0x98, 0xd4, 0xb4, 0xc9, 0x0d, 0x1c, 0xdb, 0x5d,
+	0x41, 0x9b, 0x81, 0xf7, 0xbf, 0xcc, 0x7a, 0x74, 0xf2, 0xdd, 0x02, 0x3f, 0x36, 0x5d, 0x32, 0x87,
+	0x8e, 0xf1, 0x88, 0x5c, 0xed, 0xae, 0x6e, 0x7c, 0xb8, 0x61, 0xb0, 0x7f, 0xc0, 0x9e, 0xdb, 0x50,
+	0x38, 0x63, 0x4d, 0x1d, 0x6e, 0xc3, 0xfa, 0x3a, 0xdc, 0x96, 0xab, 0x1a, 0x67, 0x1c, 0xa9, 0xc3,
+	0x6d, 0x38, 0x5e, 0x87, 0xdb, 0x32, 0xb3, 0x41, 0xee, 0xa1, 0xad, 0xfc, 0x23, 0x17, 0xbb, 0xb3,
+	0x7f, 0x6c, 0x1e, 0x5e, 0xee, 0x6b, 0x3b, 0xd0, 0x4b, 0x47, 0xff, 0xb5, 0xd3, 0x9f, 0x00, 0x00,
+	0x00, 0xff, 0xff, 0x69, 0x49, 0x0f, 0xe1, 0xeb, 0x03, 0x00, 0x00,
+}
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ context.Context
+var _ grpc.ClientConn
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+const _ = grpc.SupportPackageIsVersion4
+
+// RuntimeClient is the client API for Runtime service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream.
+type RuntimeClient interface {
+	Create(ctx context.Context, in *CreateRequest, opts ...grpc.CallOption) (*CreateResponse, error)
+	Delete(ctx context.Context, in *DeleteRequest, opts ...grpc.CallOption) (*DeleteResponse, error)
+	Update(ctx context.Context, in *UpdateRequest, opts ...grpc.CallOption) (*UpdateResponse, error)
+	List(ctx context.Context, in *ListRequest, opts ...grpc.CallOption) (*ListResponse, error)
+}
+
+type runtimeClient struct {
+	cc *grpc.ClientConn
+}
+
+func NewRuntimeClient(cc *grpc.ClientConn) RuntimeClient {
+	return &runtimeClient{cc}
+}
+
+func (c *runtimeClient) Create(ctx context.Context, in *CreateRequest, opts ...grpc.CallOption) (*CreateResponse, error) {
+	out := new(CreateResponse)
+	err := c.cc.Invoke(ctx, "/go.micro.runtime.Runtime/Create", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *runtimeClient) Delete(ctx context.Context, in *DeleteRequest, opts ...grpc.CallOption) (*DeleteResponse, error) {
+	out := new(DeleteResponse)
+	err := c.cc.Invoke(ctx, "/go.micro.runtime.Runtime/Delete", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *runtimeClient) Update(ctx context.Context, in *UpdateRequest, opts ...grpc.CallOption) (*UpdateResponse, error) {
+	out := new(UpdateResponse)
+	err := c.cc.Invoke(ctx, "/go.micro.runtime.Runtime/Update", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *runtimeClient) List(ctx context.Context, in *ListRequest, opts ...grpc.CallOption) (*ListResponse, error) {
+	out := new(ListResponse)
+	err := c.cc.Invoke(ctx, "/go.micro.runtime.Runtime/List", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// RuntimeServer is the server API for Runtime service.
+type RuntimeServer interface {
+	Create(context.Context, *CreateRequest) (*CreateResponse, error)
+	Delete(context.Context, *DeleteRequest) (*DeleteResponse, error)
+	Update(context.Context, *UpdateRequest) (*UpdateResponse, error)
+	List(context.Context, *ListRequest) (*ListResponse, error)
+}
+
+func RegisterRuntimeServer(s *grpc.Server, srv RuntimeServer) {
+	s.RegisterService(&_Runtime_serviceDesc, srv)
+}
+
+func _Runtime_Create_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(CreateRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(RuntimeServer).Create(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/go.micro.runtime.Runtime/Create",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(RuntimeServer).Create(ctx, req.(*CreateRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Runtime_Delete_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(DeleteRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(RuntimeServer).Delete(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/go.micro.runtime.Runtime/Delete",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(RuntimeServer).Delete(ctx, req.(*DeleteRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Runtime_Update_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(UpdateRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(RuntimeServer).Update(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/go.micro.runtime.Runtime/Update",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(RuntimeServer).Update(ctx, req.(*UpdateRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Runtime_List_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ListRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(RuntimeServer).List(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/go.micro.runtime.Runtime/List",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(RuntimeServer).List(ctx, req.(*ListRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+var _Runtime_serviceDesc = grpc.ServiceDesc{
+	ServiceName: "go.micro.runtime.Runtime",
+	HandlerType: (*RuntimeServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "Create",
+			Handler:    _Runtime_Create_Handler,
+		},
+		{
+			MethodName: "Delete",
+			Handler:    _Runtime_Delete_Handler,
+		},
+		{
+			MethodName: "Update",
+			Handler:    _Runtime_Update_Handler,
+		},
+		{
+			MethodName: "List",
+			Handler:    _Runtime_List_Handler,
+		},
+	},
+	Streams:  []grpc.StreamDesc{},
+	Metadata: "micro/go-micro/runtime/service/proto/runtime.proto",
+}

runtime/service/proto/runtime.proto

@@ -0,0 +1,57 @@
+syntax = "proto3";
+
+package go.micro.runtime;
+
+service Runtime {
+	rpc Create(CreateRequest) returns (CreateResponse) {};
+	rpc Delete(DeleteRequest) returns (DeleteResponse) {};
+	rpc Update(UpdateRequest) returns (UpdateResponse) {};
+	rpc List(ListRequest) returns (ListResponse) {};
+}
+
+message Service {
+	// name of the service
+	string name = 1;
+	// version of the service
+	string version = 2;
+	// git url of the source
+	string source = 3;
+	// local path of the source
+	string path = 4;
+	// command to execute
+	string exec = 5;
+}
+
+message Options {
+	// command to pass in
+	string command = 1;
+	// environment to pass in
+	repeated string env = 2;
+	// output to send to
+	string output = 3;
+}
+
+message CreateRequest {
+	Service service = 1;
+	Options options = 2;
+}
+
+message CreateResponse {}
+
+message DeleteRequest {
+	Service service = 1;
+}
+
+message DeleteResponse {}
+
+message UpdateRequest {
+	Service service = 1;
+}
+
+message UpdateResponse {}
+
+message ListRequest {}
+
+message ListResponse {
+	repeated Service services = 1;
+}

server/grpc/grpc.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"net"
 	"reflect"
+	"runtime/debug"
 	"sort"
 	"strconv"
 	"strings"
@@ -340,7 +341,8 @@ func (g *grpcServer) processRequest(stream grpc.ServerStream, service *service,
 		fn := func(ctx context.Context, req server.Request, rsp interface{}) error {
 			defer func() {
 				if r := recover(); r != nil {
-					log.Logf("handler %s panic recovered, err: %s", mtype.method.Name, r)
+					log.Log("panic recovered: ", r)
+					log.Logf(string(debug.Stack()))
 				}
 			}()
 			returnValues = function.Call([]reflect.Value{service.rcvr, mtype.prepareContext(ctx), reflect.ValueOf(argv.Interface()), reflect.ValueOf(rsp)})
@@ -621,7 +623,7 @@ func (g *grpcServer) Register() error {
 
 	g.registered = true
 
-	for sb, _ := range g.subscribers {
+	for sb := range g.subscribers {
 		handler := g.createSubHandler(sb, g.opts)
 		var opts []broker.SubscribeOption
 		if queue := sb.Options().Queue; len(queue) > 0 {

server/rpc_server.go

@@ -531,7 +531,7 @@ func (s *rpcServer) Register() error {
 
 	s.registered = true
 
-	for sb, _ := range s.subscribers {
+	for sb := range s.subscribers {
 		handler := s.createSubHandler(sb, s.opts)
 		var opts []broker.SubscribeOption
 		if queue := sb.Options().Queue; len(queue) > 0 {

store/options.go

@@ -4,7 +4,7 @@ import (
 	"github.com/micro/go-micro/config/options"
 )
 
-// Set the nodes used to back the store
+// Nodes is a list of nodes used to back the store
 func Nodes(a ...string) options.Option {
 	return options.WithValue("store.nodes", a)
 }
@@ -13,3 +13,9 @@ func Nodes(a ...string) options.Option {
 func Prefix(p string) options.Option {
 	return options.WithValue("store.prefix", p)
 }
+
+// Namespace offers a way to have multiple isolated
+// stores in the same backend, if supported.
+func Namespace(n string) options.Option {
+	return options.WithValue("store.namespace", n)
+}

store/postgresql/postgresql.go

@@ -0,0 +1,248 @@
+// Package postgresql implements a micro Store backed by sql
+package postgresql
+
+import (
+	"database/sql"
+	"fmt"
+	"strings"
+	"time"
+	"unicode"
+
+	"github.com/lib/pq"
+	"github.com/pkg/errors"
+
+	"github.com/micro/go-micro/config/options"
+	"github.com/micro/go-micro/store"
+)
+
+// DefaultNamespace is the namespace that the sql store
+// will use if no namespace is provided.
+const DefaultNamespace = "micro"
+
+type sqlStore struct {
+	db *sql.DB
+
+	table string
+	options.Options
+}
+
+// List all the known records
+func (s *sqlStore) List() ([]*store.Record, error) {
+	q, err := s.db.Prepare(fmt.Sprintf("SELECT key, value, expiry FROM micro.%s;", s.table))
+	if err != nil {
+		return nil, err
+	}
+	var records []*store.Record
+	var timehelper pq.NullTime
+	rows, err := q.Query()
+	if err != nil {
+		if err == sql.ErrNoRows {
+			return records, nil
+		}
+		return nil, err
+	}
+	defer rows.Close()
+	for rows.Next() {
+		record := &store.Record{}
+		if err := rows.Scan(&record.Key, &record.Value, &timehelper); err != nil {
+			return records, err
+		}
+		if timehelper.Valid {
+			if timehelper.Time.Before(time.Now()) {
+				// record has expired
+				go s.Delete(record.Key)
+			} else {
+				record.Expiry = time.Until(timehelper.Time)
+				records = append(records, record)
+			}
+		} else {
+			records = append(records, record)
+		}
+
+	}
+	rowErr := rows.Close()
+	if rowErr != nil {
+		// transaction rollback or something
+		return records, rowErr
+	}
+	if err := rows.Err(); err != nil {
+		return records, err
+	}
+	return records, nil
+}
+
+// Read all records with keys
+func (s *sqlStore) Read(keys ...string) ([]*store.Record, error) {
+	q, err := s.db.Prepare(fmt.Sprintf("SELECT key, value, expiry FROM micro.%s WHERE key = $1;", s.table))
+	if err != nil {
+		return nil, err
+	}
+	var records []*store.Record
+	var timehelper pq.NullTime
+	for _, key := range keys {
+		row := q.QueryRow(key)
+		record := &store.Record{}
+		if err := row.Scan(&record.Key, &record.Value, &timehelper); err != nil {
+			if err == sql.ErrNoRows {
+				return records, store.ErrNotFound
+			}
+			return records, err
+		}
+		if timehelper.Valid {
+			if timehelper.Time.Before(time.Now()) {
+				// record has expired
+				go s.Delete(key)
+				return records, store.ErrNotFound
+			}
+			record.Expiry = time.Until(timehelper.Time)
+			records = append(records, record)
+		} else {
+			records = append(records, record)
+		}
+	}
+	return records, nil
+}
+
+// Write records
+func (s *sqlStore) Write(rec ...*store.Record) error {
+	q, err := s.db.Prepare(fmt.Sprintf(`INSERT INTO micro.%s(key, value, expiry)
+		VALUES ($1, $2::bytea, $3)
+		ON CONFLICT (key)
+		DO UPDATE
+		SET value = EXCLUDED.value, expiry = EXCLUDED.expiry;`, s.table))
+	if err != nil {
+		return err
+	}
+	for _, r := range rec {
+		var err error
+		if r.Expiry != 0 {
+			_, err = q.Exec(r.Key, r.Value, time.Now().Add(r.Expiry))
+		} else {
+			_, err = q.Exec(r.Key, r.Value, nil)
+		}
+		if err != nil {
+			return errors.Wrap(err, "Couldn't insert record "+r.Key)
+		}
+	}
+
+	return nil
+}
+
+// Delete records with keys
+func (s *sqlStore) Delete(keys ...string) error {
+	q, err := s.db.Prepare(fmt.Sprintf("DELETE FROM micro.%s WHERE key = $1;", s.table))
+	if err != nil {
+		return err
+	}
+	for _, key := range keys {
+		result, err := q.Exec(key)
+		if err != nil {
+			return err
+		}
+		_, err = result.RowsAffected()
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (s *sqlStore) initDB(options options.Options) error {
+	// Get the store.namespace option, or use sql.DefaultNamespace
+	namespaceOpt, found := options.Values().Get("store.namespace")
+	if !found {
+		s.table = DefaultNamespace
+	} else {
+		if namespace, ok := namespaceOpt.(string); ok {
+			s.table = namespace
+		} else {
+			return errors.New("store.namespace option must be a string")
+		}
+	}
+
+	// Create "micro" schema
+	schema, err := s.db.Prepare("CREATE SCHEMA IF NOT EXISTS micro ;")
+	if err != nil {
+		return err
+	}
+	_, err = schema.Exec()
+	if err != nil {
+		return errors.Wrap(err, "Couldn't create Schema")
+	}
+
+	// Create a table for the Store namespace
+	tableq, err := s.db.Prepare(fmt.Sprintf(`CREATE TABLE IF NOT EXISTS micro.%s
+	(
+		key text COLLATE "default" NOT NULL,
+		value bytea,
+		expiry timestamp with time zone,
+		CONSTRAINT %s_pkey PRIMARY KEY (key)
+	);`, s.table, s.table))
+	_, err = tableq.Exec()
+	if err != nil {
+		return errors.Wrap(err, "Couldn't create table")
+	}
+
+	return nil
+}
+
+// New returns a new micro Store backed by sql
+func New(opts ...options.Option) (store.Store, error) {
+	options := options.NewOptions(opts...)
+	driver, dataSourceName, err := validateOptions(options)
+	if err != nil {
+		return nil, err
+	}
+	if !strings.Contains(dataSourceName, " ") {
+		dataSourceName = fmt.Sprintf("host=%s", dataSourceName)
+	}
+	db, err := sql.Open(driver, dataSourceName)
+	if err != nil {
+		return nil, err
+	}
+	if err := db.Ping(); err != nil {
+		return nil, err
+	}
+	s := &sqlStore{
+		db: db,
+	}
+
+	return s, s.initDB(options)
+}
+
+// validateOptions checks whether the provided options are valid, then returns the driver
+// and data source name.
+func validateOptions(options options.Options) (driver, dataSourceName string, err error) {
+	driverOpt, found := options.Values().Get("store.sql.driver")
+	if !found {
+		return "", "", errors.New("No store.sql.driver option specified")
+	}
+	nodesOpt, found := options.Values().Get("store.nodes")
+	if !found {
+		return "", "", errors.New("No store.nodes option specified (expected a database connection string)")
+	}
+	driver, ok := driverOpt.(string)
+	if !ok {
+		return "", "", errors.New("store.sql.driver option must be a string")
+	}
+	nodes, ok := nodesOpt.([]string)
+	if !ok {
+		return "", "", errors.New("store.nodes option must be a []string")
+	}
+	if len(nodes) != 1 {
+		return "", "", errors.New("expected only 1 store.nodes option")
+	}
+	namespaceOpt, found := options.Values().Get("store.namespace")
+	if found {
+		namespace, ok := namespaceOpt.(string)
+		if !ok {
+			return "", "", errors.New("store.namespace must me a string")
+		}
+		for _, r := range namespace {
+			if !unicode.IsLetter(r) {
+				return "", "", errors.New("store.namespace must only contain letters")
+			}
+		}
+	}
+	return driver, nodes[0], nil
+}

store/postgresql/postgresql_test.go

@@ -0,0 +1,95 @@
+package postgresql
+
+import (
+	"database/sql"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/kr/pretty"
+	"github.com/micro/go-micro/store"
+)
+
+func TestSQL(t *testing.T) {
+	connection := fmt.Sprintf(
+		"host=%s port=%d user=%s sslmode=disable dbname=%s",
+		"localhost",
+		5432,
+		"jake",
+		"test",
+	)
+	db, err := sql.Open("postgres", connection)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Ping(); err != nil {
+		t.Skip(err)
+	}
+	db.Close()
+
+	sqlStore, err := New(
+		store.Namespace("testsql"),
+		store.Nodes(connection),
+	)
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+
+	records, err := sqlStore.List()
+	if err != nil {
+		t.Error(err)
+	} else {
+		t.Logf("%# v\n", pretty.Formatter(records))
+	}
+
+	err = sqlStore.Write(
+		&store.Record{
+			Key:   "test",
+			Value: []byte("foo"),
+		},
+		&store.Record{
+			Key:   "bar",
+			Value: []byte("baz"),
+		},
+		&store.Record{
+			Key:   "qux",
+			Value: []byte("aasad"),
+		},
+	)
+	if err != nil {
+		t.Error(err)
+	}
+	err = sqlStore.Delete("qux")
+	if err != nil {
+		t.Error(err)
+	}
+
+	err = sqlStore.Write(&store.Record{
+		Key:    "test",
+		Value:  []byte("bar"),
+		Expiry: time.Minute,
+	})
+	if err != nil {
+		t.Error(err)
+	}
+
+	records, err = sqlStore.Read("test")
+	if err != nil {
+		t.Error(err)
+	} else {
+		t.Logf("%# v\n", pretty.Formatter(records))
+		if string(records[0].Value) != "bar" {
+			t.Error("Expected bar, got ", string(records[0].Value))
+		}
+	}
+
+	time.Sleep(61 * time.Second)
+	records, err = sqlStore.Read("test")
+	if err == nil {
+		t.Error("Key test should have expired")
+	} else {
+		if err != store.ErrNotFound {
+			t.Error(err)
+		}
+	}
+}

store/store.go

@@ -7,6 +7,7 @@ import (
 )
 
 var (
+	// ErrNotFound is returned when a Read key doesn't exist
 	ErrNotFound = errors.New("not found")
 )
 
@@ -14,11 +15,11 @@ var (
 type Store interface {
 	// List all the known records
 	List() ([]*Record, error)
-	// Read a record with key
+	// Read records with keys
 	Read(key ...string) ([]*Record, error)
-	// Write a record
+	// Write records
 	Write(rec ...*Record) error
-	// Delete a record with key
+	// Delete records with keys
 	Delete(key ...string) error
 }
 

sync/lock/memory/memory.go

@@ -1,4 +1,4 @@
-// Package memoy provides a sync.Mutex implementation of the lock for local use
+// Package memory provides a sync.Mutex implementation of the lock for local use
 package memory
 
 import (

sync/map.go

@@ -5,6 +5,7 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"sort"
 
 	"github.com/micro/go-micro/store"
 	ckv "github.com/micro/go-micro/store/etcd"
@@ -94,6 +95,10 @@ func (m *syncMap) Iterate(fn func(key, val interface{}) error) error {
 		return err
 	}
 
+	sort.Slice(keyvals, func(i, j int) bool {
+		return keyvals[i].Key < keyvals[j].Key
+	})
+
 	for _, keyval := range keyvals {
 		// lock
 		if err := m.opts.Lock.Acquire(keyval.Key); err != nil {

sync/map_test.go

@@ -0,0 +1,39 @@
+package sync
+
+import (
+	"testing"
+	"time"
+
+	store "github.com/micro/go-micro/store"
+	mem_store "github.com/micro/go-micro/store/memory"
+	mem_lock "github.com/micro/go-micro/sync/lock/memory"
+)
+
+func TestIterate(t *testing.T) {
+	s1 := mem_store.NewStore()
+	s2 := mem_store.NewStore()
+	recA := &store.Record{
+		Key:   "A",
+		Value: nil,
+	}
+	recB := &store.Record{
+		Key:   "B",
+		Value: nil,
+	}
+	s1.Write(recA)
+	s1.Write(recB)
+	s2.Write(recB)
+	s2.Write(recA)
+
+	f := func(key, val interface{}) error {
+		time.Sleep(1 * time.Millisecond)
+		return nil
+	}
+	l := mem_lock.NewLock()
+	m1 := NewMap(WithStore(s1), WithLock(l))
+	m2 := NewMap(WithStore(s2), WithLock(l))
+	go func() {
+		m2.Iterate(f)
+	}()
+	m1.Iterate(f)
+}

tunnel/default.go

@@ -225,9 +225,10 @@ func (t *tun) monitor() {
 				t.Lock()
 				for _, node := range delLinks {
 					log.Debugf("Tunnel deleting dead link for %s", node)
-					link := t.links[node]
-					link.Close()
-					delete(t.links, node)
+					if link, ok := t.links[node]; ok {
+						link.Close()
+						delete(t.links, node)
+					}
 				}
 				t.Unlock()
 			}
@@ -362,7 +363,7 @@ func (t *tun) process() {
 			// send the message
 			for _, link := range sendTo {
 				// send the message via the current link
-				log.Debugf("Sending %+v to %s", newMsg.Header, link.Remote())
+				log.Tracef("Sending %+v to %s", newMsg.Header, link.Remote())
 
 				if errr := link.Send(newMsg); errr != nil {
 					log.Debugf("Tunnel error sending %+v to %s: %v", newMsg.Header, link.Remote(), errr)
@@ -547,7 +548,7 @@ func (t *tun) listen(link *link) {
 		// a continued session
 		case "session":
 			// process message
-			log.Debugf("Received %+v from %s", msg.Header, link.Remote())
+			log.Tracef("Received %+v from %s", msg.Header, link.Remote())
 		// an announcement of a channel listener
 		case "announce":
 			// process the announcement
@@ -589,7 +590,7 @@ func (t *tun) listen(link *link) {
 		}
 
 		// strip tunnel message header
-		for k, _ := range msg.Header {
+		for k := range msg.Header {
 			if strings.HasPrefix(k, "Micro-Tunnel") {
 				delete(msg.Header, k)
 			}

tunnel/session.go

@@ -321,7 +321,7 @@ func (s *session) Send(m *transport.Message) error {
 		msg.link = ""
 	}
 
-	log.Debugf("Appending %+v to send backlog", msg)
+	log.Tracef("Appending %+v to send backlog", msg)
 	// send the actual message
 	s.send <- msg
 
@@ -352,7 +352,7 @@ func (s *session) Recv(m *transport.Message) error {
 	default:
 	}
 
-	log.Debugf("Received %+v from recv backlog", msg)
+	log.Tracef("Received %+v from recv backlog", msg)
 	// set message
 	*m = *msg.data
 	// return nil

web/service.go

@@ -82,7 +82,7 @@ func (s *service) genSrv() *registry.Service {
 	return &registry.Service{
 		Name:    s.opts.Name,
 		Version: s.opts.Version,
-		Nodes: []*registry.Node{&registry.Node{
+		Nodes: []*registry.Node{{
 			Id:       s.opts.Id,
 			Address:  fmt.Sprintf("%s:%d", addr, port),
 			Metadata: s.opts.Metadata,
@@ -118,6 +118,11 @@ func (s *service) register() error {
 	if s.opts.Registry != nil {
 		r = s.opts.Registry
 	}
+
+	// service node need modify, node address maybe changed
+	srv := s.genSrv()
+	srv.Endpoints = s.srv.Endpoints
+	s.srv = srv
 	return r.Register(s.srv, registry.RegisterTTL(s.opts.RegisterTTL))
 }