Repository: weaveworks/flux (go)

Issue:
Add Upstream API to regular API
At present, there are a small number of methods in the Upstream API that are not in the flux API. These are methods that can be called via RPC from an upstream (i.e., Weave Cloud) connection, but not via HTTP (e.g., not by `fluxctl`).

In the course of separating the Upstream part of the daemon out into weaveworks/flux-adapter, these will have to move into the public API, since `flux-adapter` runs in a sidecar, and uses HTTP (via http/client) rather than having in-process access to the daemon.

I believe the upstream methods can just be added to the v11 Server interface, and the Upstream interface abandoned from now on.

Pull request:
Flux Adapter preparation
 - makes API changes so that all methods are available over HTTP, including NotifyChange (used for relaying webhooks, for example) /cc @alanjcastonguay 
 - adds some notes about what can be moved out of this repo and into `weaveworks/flux-adapter`

This (modulo passing tests and review) can be merged now, in principle, because it doesn't alter what is served over the Upstream connection -- it just adds to the HTTP API. But I am going to make it a draft until I've Actually Tried that.

Fixes #2210. 

Patch:
diff --git a/api/api.go b/api/api.go
index a5fd7273..6e03a1ab 100644
--- a/api/api.go
+++ b/api/api.go
@@ -8,10 +8,3 @@ import "github.com/weaveworks/flux/api/v11"
 type Server interface {
 	v11.Server
 }
-
-// UpstreamServer is the interface a Flux must satisfy in order to communicate with
-// Weave Cloud.
-type UpstreamServer interface {
-	v11.Server
-	v11.Upstream
-}
diff --git a/api/v11/api.go b/api/v11/api.go
index 5b591b9a..d8f23b3e 100644
--- a/api/v11/api.go
+++ b/api/v11/api.go
@@ -18,8 +18,8 @@ type Server interface {
 	v10.Server
 
 	ListServicesWithOptions(ctx context.Context, opts ListServicesOptions) ([]v6.ControllerStatus, error)
-}
 
-type Upstream interface {
+	// NB Upstream methods move into the public API, since
+	// weaveworks/flux-adapter now relies on the public API
 	v10.Upstream
 }
diff --git a/cmd/fluxd/main.go b/cmd/fluxd/main.go
index 9ceadf20..5d558767 100644
--- a/cmd/fluxd/main.go
+++ b/cmd/fluxd/main.go
@@ -609,7 +609,7 @@ func main() {
 				client.Token(*token),
 				transport.NewUpstreamRouter(),
 				*upstreamURL,
-				remote.NewErrorLoggingUpstreamServer(daemon, upstreamLogger),
+				remote.NewErrorLoggingServer(daemon, upstreamLogger),
 				*rpcTimeout,
 				upstreamLogger,
 			)
diff --git a/http/client/client.go b/http/client/client.go
index 3fe4d8c1..9a937d72 100644
--- a/http/client/client.go
+++ b/http/client/client.go
@@ -16,6 +16,7 @@ import (
 	"github.com/weaveworks/flux/api/v10"
 	"github.com/weaveworks/flux/api/v11"
 	"github.com/weaveworks/flux/api/v6"
+	"github.com/weaveworks/flux/api/v9"
 	fluxerr "github.com/weaveworks/flux/errors"
 	"github.com/weaveworks/flux/event"
 	transport "github.com/weaveworks/flux/http"
@@ -53,6 +54,20 @@ func New(c *http.Client, router *mux.Router, endpoint string, t Token) *Client {
 	}
 }
 
+func (c *Client) Ping(ctx context.Context) error {
+	return c.Get(ctx, nil, transport.Ping)
+}
+
+func (c *Client) Version(ctx context.Context) (string, error) {
+	var v string
+	err := c.Get(ctx, &v, transport.Version)
+	return v, err
+}
+
+func (c *Client) NotifyChange(ctx context.Context, change v9.Change) error {
+	return c.PostWithBody(ctx, transport.Notify, change)
+}
+
 func (c *Client) ListServices(ctx context.Context, namespace string) ([]v6.ControllerStatus, error) {
 	var res []v6.ControllerStatus
 	err := c.Get(ctx, &res, transport.ListServices, "namespace", namespace)
@@ -117,7 +132,7 @@ func (c *Client) GitRepoConfig(ctx context.Context, regenerate bool) (v6.GitConf
 
 // --- Request helpers
 
-// post is a simple query-param only post request
+// Post is a simple query-param only post request
 func (c *Client) Post(ctx context.Context, route string, queryParams ...string) error {
 	return c.PostWithBody(ctx, route, nil, queryParams...)
 }
@@ -178,7 +193,7 @@ func (c *Client) methodWithResp(ctx context.Context, method string, dest interfa
 	return nil
 }
 
-// get executes a get request against the Flux server. it unmarshals the response into dest.
+// Get executes a get request against the Flux server. it unmarshals the response into dest, if not nil.
 func (c *Client) Get(ctx context.Context, dest interface{}, route string, queryParams ...string) error {
 	u, err := transport.MakeURL(c.endpoint, c.router, route, queryParams...)
 	if err != nil {
@@ -200,8 +215,10 @@ func (c *Client) Get(ctx context.Context, dest interface{}, route string, queryP
 	}
 	defer resp.Body.Close()
 
-	if err := json.NewDecoder(resp.Body).Decode(dest); err != nil {
-		return errors.Wrap(err, "decoding response from server")
+	if dest != nil {
+		if err := json.NewDecoder(resp.Body).Decode(dest); err != nil {
+			return errors.Wrap(err, "decoding response from server")
+		}
 	}
 	return nil
 }
@@ -212,7 +229,7 @@ func (c *Client) executeRequest(req *http.Request) (*http.Response, error) {
 		return nil, errors.Wrap(err, "executing HTTP request")
 	}
 	switch resp.StatusCode {
-	case http.StatusOK, http.StatusCreated, http.StatusNoContent:
+	case http.StatusOK, http.StatusCreated, http.StatusNoContent, http.StatusAccepted:
 		return resp, nil
 	case http.StatusUnauthorized:
 		return resp, transport.ErrorUnauthorized
diff --git a/http/daemon/server.go b/http/daemon/server.go
index a5093385..5abd4007 100644
--- a/http/daemon/server.go
+++ b/http/daemon/server.go
@@ -13,6 +13,7 @@ import (
 	"github.com/weaveworks/flux/api"
 	"github.com/weaveworks/flux/api/v10"
 	"github.com/weaveworks/flux/api/v11"
+	"github.com/weaveworks/flux/api/v9"
 	transport "github.com/weaveworks/flux/http"
 	"github.com/weaveworks/flux/job"
 	fluxmetrics "github.com/weaveworks/flux/metrics"
@@ -47,6 +48,13 @@ func NewRouter() *mux.Router {
 
 func NewHandler(s api.Server, r *mux.Router) http.Handler {
 	handle := HTTPServer{s}
+
+	// Erstwhile Upstream(Server) methods, now part of v11
+	r.Get(transport.Ping).HandlerFunc(handle.Ping)
+	r.Get(transport.Version).HandlerFunc(handle.Version)
+	r.Get(transport.Notify).HandlerFunc(handle.Notify)
+
+	// v6-v11 handlers
 	r.Get(transport.ListServices).HandlerFunc(handle.ListServicesWithOptions)
 	r.Get(transport.ListServicesWithOptions).HandlerFunc(handle.ListServicesWithOptions)
 	r.Get(transport.ListImages).HandlerFunc(handle.ListImagesWithOptions)
@@ -74,6 +82,39 @@ type HTTPServer struct {
 	server api.Server
 }
 
+func (s HTTPServer) Ping(w http.ResponseWriter, r *http.Request) {
+	if err := s.server.Ping(r.Context()); err != nil {
+		transport.ErrorResponse(w, r, err)
+		return
+	}
+	w.WriteHeader(http.StatusNoContent)
+	return
+}
+
+func (s HTTPServer) Version(w http.ResponseWriter, r *http.Request) {
+	version, err := s.server.Version(r.Context())
+	if err != nil {
+		transport.ErrorResponse(w, r, err)
+		return
+	}
+	transport.JSONResponse(w, r, version)
+}
+
+func (s HTTPServer) Notify(w http.ResponseWriter, r *http.Request) {
+	var change v9.Change
+	defer r.Body.Close()
+
+	if err := json.NewDecoder(r.Body).Decode(&change); err != nil {
+		transport.WriteError(w, r, http.StatusBadRequest, err)
+		return
+	}
+	if err := s.server.NotifyChange(r.Context(), change); err != nil {
+		transport.ErrorResponse(w, r, err)
+		return
+	}
+	w.WriteHeader(http.StatusAccepted)
+}
+
 func (s HTTPServer) JobStatus(w http.ResponseWriter, r *http.Request) {
 	id := job.ID(mux.Vars(r)["id"])
 	status, err := s.server.JobStatus(r.Context(), id)
diff --git a/http/daemon/upstream.go b/http/daemon/upstream.go
index cdea3e43..3ad2f075 100644
--- a/http/daemon/upstream.go
+++ b/http/daemon/upstream.go
@@ -1,5 +1,8 @@
 package daemon
 
+// This file can be removed from the package once `--connect` is
+// removed from fluxd. Until then, it will be imported from here.
+
 import (
 	"context"
 	"net/http"
@@ -29,7 +32,7 @@ type Upstream struct {
 	url       *url.URL
 	endpoint  string
 	apiClient *fluxclient.Client
-	server    api.UpstreamServer
+	server    api.Server
 	timeout   time.Duration
 	logger    log.Logger
 	quit      chan struct{}
@@ -47,13 +50,13 @@ var (
 	}, []string{"target"})
 )
 
-func NewUpstream(client *http.Client, ua string, t fluxclient.Token, router *mux.Router, endpoint string, s api.UpstreamServer, timeout time.Duration, logger log.Logger) (*Upstream, error) {
+func NewUpstream(client *http.Client, ua string, t fluxclient.Token, router *mux.Router, endpoint string, s api.Server, timeout time.Duration, logger log.Logger) (*Upstream, error) {
 	httpEndpoint, wsEndpoint, err := inferEndpoints(endpoint)
 	if err != nil {
 		return nil, errors.Wrap(err, "inferring WS/HTTP endpoints")
 	}
 
-	u, err := transport.MakeURL(wsEndpoint, router, transport.RegisterDaemonV10)
+	u, err := transport.MakeURL(wsEndpoint, router, transport.RegisterDaemonV11)
 	if err != nil {
 		return nil, errors.Wrap(err, "constructing URL")
 	}
diff --git a/http/routes.go b/http/routes.go
index 58e9b9cb..3d062e7e 100644
--- a/http/routes.go
+++ b/http/routes.go
@@ -1,6 +1,11 @@
 package http
 
 const (
+	// Formerly Upstream methods, now (in v11) included in server API
+	Ping    = "Ping"
+	Version = "Version"
+	Notify  = "Notify"
+
 	ListServices            = "ListServices"
 	ListServicesWithOptions = "ListServicesWithOptions"
 	ListImages              = "ListImages"
@@ -17,12 +22,20 @@ const (
 	RegeneratePublicSSHKey = "RegeneratePublicSSHKey"
 )
 
-var (
+// This is part of the API -- but it's the outward-facing (or service
+// provider) API, rather than the flux API.
+const (
+	LogEvent = "LogEvent"
+)
+
+// The RegisterDaemonX routes should move to weaveworks/flux-adapter
+// once we remove `--connect`, since they are pertinent only to making
+// an RPC relay connection.
+const (
 	RegisterDaemonV6  = "RegisterDaemonV6"
 	RegisterDaemonV7  = "RegisterDaemonV7"
 	RegisterDaemonV8  = "RegisterDaemonV8"
 	RegisterDaemonV9  = "RegisterDaemonV9"
 	RegisterDaemonV10 = "RegisterDaemonV10"
 	RegisterDaemonV11 = "RegisterDaemonV11"
-	LogEvent          = "LogEvent"
 )
diff --git a/http/transport.go b/http/transport.go
index faee6b6a..bbe5552b 100644
--- a/http/transport.go
+++ b/http/transport.go
@@ -29,6 +29,10 @@ func DeprecateVersions(r *mux.Router, versions ...string) {
 func NewAPIRouter() *mux.Router {
 	r := mux.NewRouter()
 
+	r.NewRoute().Name(Ping).Methods("GET").Path("/v11/ping")
+	r.NewRoute().Name(Version).Methods("GET").Path("/v11/version")
+	r.NewRoute().Name(Notify).Methods("POST").Path("/v11/notify")
+
 	r.NewRoute().Name(ListServices).Methods("GET").Path("/v6/services")
 	r.NewRoute().Name(ListServicesWithOptions).Methods("GET").Path("/v11/services")
 	r.NewRoute().Name(ListImages).Methods("GET").Path("/v6/images")
@@ -50,6 +54,8 @@ func NewAPIRouter() *mux.Router {
 	return r // TODO 404 though?
 }
 
+// These (routes, and constructor following) should move to
+// weaveworks/flux-adapter when `--connect` is removed from fluxd.
 func UpstreamRoutes(r *mux.Router) {
 	r.NewRoute().Name(RegisterDaemonV6).Methods("GET").Path("/v6/daemon")
 	r.NewRoute().Name(RegisterDaemonV7).Methods("GET").Path("/v7/daemon")
diff --git a/http/websocket/websocket.go b/http/websocket/websocket.go
index 5a60f2ee..f9bd5b8a 100644
--- a/http/websocket/websocket.go
+++ b/http/websocket/websocket.go
@@ -1,5 +1,10 @@
 package websocket
 
+// This package can be moved to weaveworks/flux-adapter once
+// `--connect` is removed, since it is particular to making an RPC
+// relay connection, and that function will be supplied by
+// flux-adapter.
+
 import (
 	"io"
 
diff --git a/remote/doc.go b/remote/doc.go
index 5d946bf0..dedd84af 100644
--- a/remote/doc.go
+++ b/remote/doc.go
@@ -2,3 +2,7 @@
 // an upstream service.
 
 package remote
+
+// This whole package can be moved to weaveworks/flux-adapter, once we
+// no longer `--connect` from fluxd. (Or before that, if we import it
+// from there.)
diff --git a/remote/logging.go b/remote/logging.go
index e1844291..4fb21797 100644
--- a/remote/logging.go
+++ b/remote/logging.go
@@ -15,7 +15,6 @@ import (
 )
 
 var _ api.Server = &ErrorLoggingServer{}
-var _ api.UpstreamServer = &ErrorLoggingUpstreamServer{}
 
 type ErrorLoggingServer struct {
 	server api.Server
@@ -108,19 +107,7 @@ func (p *ErrorLoggingServer) GitRepoConfig(ctx context.Context, regenerate bool)
 	return p.server.GitRepoConfig(ctx, regenerate)
 }
 
-type ErrorLoggingUpstreamServer struct {
-	*ErrorLoggingServer
-	server api.UpstreamServer
-}
-
-func NewErrorLoggingUpstreamServer(s api.UpstreamServer, l log.Logger) *ErrorLoggingUpstreamServer {
-	return &ErrorLoggingUpstreamServer{
-		NewErrorLoggingServer(s, l),
-		s,
-	}
-}
-
-func (p *ErrorLoggingUpstreamServer) Ping(ctx context.Context) (err error) {
+func (p *ErrorLoggingServer) Ping(ctx context.Context) (err error) {
 	defer func() {
 		if err != nil {
 			p.logger.Log("method", "Ping", "error", err)
@@ -129,7 +116,7 @@ func (p *ErrorLoggingUpstreamServer) Ping(ctx context.Context) (err error) {
 	return p.server.Ping(ctx)
 }
 
-func (p *ErrorLoggingUpstreamServer) Version(ctx context.Context) (v string, err error) {
+func (p *ErrorLoggingServer) Version(ctx context.Context) (v string, err error) {
 	defer func() {
 		if err != nil {
 			p.logger.Log("method", "Version", "error", err, "version", v)
@@ -138,7 +125,7 @@ func (p *ErrorLoggingUpstreamServer) Version(ctx context.Context) (v string, err
 	return p.server.Version(ctx)
 }
 
-func (p *ErrorLoggingUpstreamServer) NotifyChange(ctx context.Context, change v9.Change) (err error) {
+func (p *ErrorLoggingServer) NotifyChange(ctx context.Context, change v9.Change) (err error) {
 	defer func() {
 		if err != nil {
 			p.logger.Log("method", "NotifyChange", "error", err)
diff --git a/remote/metrics.go b/remote/metrics.go
index 5ecf038d..eea35b3f 100644
--- a/remote/metrics.go
+++ b/remote/metrics.go
@@ -128,21 +128,7 @@ func (i *instrumentedServer) GitRepoConfig(ctx context.Context, regenerate bool)
 	return i.s.GitRepoConfig(ctx, regenerate)
 }
 
-var _ api.UpstreamServer = &instrumentedUpstreamServer{}
-
-type instrumentedUpstreamServer struct {
-	*instrumentedServer
-	s api.UpstreamServer
-}
-
-func InstrumentUpstream(s api.UpstreamServer) *instrumentedUpstreamServer {
-	return &instrumentedUpstreamServer{
-		Instrument(s),
-		s,
-	}
-}
-
-func (i *instrumentedUpstreamServer) Ping(ctx context.Context) (err error) {
+func (i *instrumentedServer) Ping(ctx context.Context) (err error) {
 	defer func(begin time.Time) {
 		requestDuration.With(
 			fluxmetrics.LabelMethod, "Ping",
@@ -152,7 +138,7 @@ func (i *instrumentedUpstreamServer) Ping(ctx context.Context) (err error) {
 	return i.s.Ping(ctx)
 }
 
-func (i *instrumentedUpstreamServer) Version(ctx context.Context) (v string, err error) {
+func (i *instrumentedServer) Version(ctx context.Context) (v string, err error) {
 	defer func(begin time.Time) {
 		requestDuration.With(
 			fluxmetrics.LabelMethod, "Version",
@@ -162,7 +148,7 @@ func (i *instrumentedUpstreamServer) Version(ctx context.Context) (v string, err
 	return i.s.Version(ctx)
 }
 
-func (i *instrumentedUpstreamServer) NotifyChange(ctx context.Context, change v9.Change) (err error) {
+func (i *instrumentedServer) NotifyChange(ctx context.Context, change v9.Change) (err error) {
 	defer func(begin time.Time) {
 		requestDuration.With(
 			fluxmetrics.LabelMethod, "NotifyChange",
diff --git a/remote/mock.go b/remote/mock.go
index 1d889172..d4573c0d 100644
--- a/remote/mock.go
+++ b/remote/mock.go
@@ -104,13 +104,13 @@ func (p *MockServer) GitRepoConfig(ctx context.Context, regenerate bool) (v6.Git
 	return p.GitRepoConfigAnswer, p.GitRepoConfigError
 }
 
-var _ api.UpstreamServer = &MockServer{}
+var _ api.Server = &MockServer{}
 
 // -- Battery of tests for an api.Server implementation. Since these
 // essentially wrap the server in various transports, we expect
 // arguments and answers to be preserved.
 
-func ServerTestBattery(t *testing.T, wrap func(mock api.UpstreamServer) api.UpstreamServer) {
+func ServerTestBattery(t *testing.T, wrap func(mock api.Server) api.Server) {
 	// set up
 	namespace := "the-space-of-names"
 	serviceID := resource.MustParseID(namespace + "/service")
diff --git a/remote/rpc/baseclient.go b/remote/rpc/baseclient.go
index 10e898c4..977336f0 100644
--- a/remote/rpc/baseclient.go
+++ b/remote/rpc/baseclient.go
@@ -17,7 +17,7 @@ import (
 
 type baseClient struct{}
 
-var _ api.UpstreamServer = baseClient{}
+var _ api.Server = baseClient{}
 
 func (bc baseClient) Version(context.Context) (string, error) {
 	return "", remote.UpgradeNeededError(errors.New("Version method not implemented"))
diff --git a/remote/rpc/clientV11.go b/remote/rpc/clientV11.go
index 2f4d063c..c3b96691 100644
--- a/remote/rpc/clientV11.go
+++ b/remote/rpc/clientV11.go
@@ -19,7 +19,6 @@ type RPCClientV11 struct {
 
 type clientV11 interface {
 	v11.Server
-	v11.Upstream
 }
 
 var _ clientV11 = &RPCClientV11{}
diff --git a/remote/rpc/server.go b/remote/rpc/server.go
index 2d9f850b..4e662340 100644
--- a/remote/rpc/server.go
+++ b/remote/rpc/server.go
@@ -26,7 +26,7 @@ type Server struct {
 
 // NewServer instantiates a new RPC server, handling requests on the
 // conn by invoking methods on the underlying (assumed local) server.
-func NewServer(s api.UpstreamServer, t time.Duration) (*Server, error) {
+func NewServer(s api.Server, t time.Duration) (*Server, error) {
 	server := rpc.NewServer()
 	if err := server.Register(&RPCServer{s, t}); err != nil {
 		return nil, err
@@ -39,7 +39,7 @@ func (c *Server) ServeConn(conn io.ReadWriteCloser) {
 }
 
 type RPCServer struct {
-	s       api.UpstreamServer
+	s       api.Server
 	timeout time.Duration
 }
 


Test patch:
diff --git a/remote/mock_test.go b/remote/mock_test.go
index e26fbdb6..d299744b 100644
--- a/remote/mock_test.go
+++ b/remote/mock_test.go
@@ -8,5 +8,5 @@ import (
 
 // Just test that the mock does its job.
 func TestMock(t *testing.T) {
-	ServerTestBattery(t, func(mock api.UpstreamServer) api.UpstreamServer { return mock })
+	ServerTestBattery(t, func(mock api.Server) api.Server { return mock })
 }
diff --git a/remote/rpc/rpc_test.go b/remote/rpc/rpc_test.go
index 88b3ef48..6033bcbd 100644
--- a/remote/rpc/rpc_test.go
+++ b/remote/rpc/rpc_test.go
@@ -24,7 +24,7 @@ func pipes() (io.ReadWriteCloser, io.ReadWriteCloser) {
 }
 
 func TestRPC(t *testing.T) {
-	wrap := func(mock api.UpstreamServer) api.UpstreamServer {
+	wrap := func(mock api.Server) api.Server {
 		clientConn, serverConn := pipes()
 
 		server, err := NewServer(mock, 10*time.Second)