Skip to content
This repository was archived by the owner on Jan 21, 2020. It is now read-only.

Use JSON-RPC over HTTP for communication between plugins #297

Merged
merged 3 commits into from
Nov 15, 2016
Merged

Use JSON-RPC over HTTP for communication between plugins #297

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6716041
Use JSON-RPC over HTTP for communication between plugins
wfarner Nov 15, 2016
99b03c7
Vendor updates
wfarner Nov 15, 2016
078728e
Merge branch 'master' into gorilla-rpc-with-logging
wfarner Nov 15, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 4 additions & 5 deletions cli/serverutil.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,16 +5,15 @@ import (

log "github.com/Sirupsen/logrus"
"github.com/docker/infrakit/discovery"
"github.com/docker/infrakit/rpc"
"github.com/docker/infrakit/rpc/server"
)

// RunPlugin runs a plugin server, advertising with the provided name for discovery.
// THe plugin should conform to the rpc call convention as implemented in the rpc package.
// The plugin should conform to the rpc call convention as implemented in the rpc package.
func RunPlugin(name string, plugin interface{}) {
_, stopped, err := rpc.StartPluginAtPath(path.Join(discovery.Dir(), name), plugin)
stoppable, err := server.StartPluginAtPath(path.Join(discovery.Dir(), name), plugin)
if err != nil {
log.Error(err)
}

<-stopped // block until done
stoppable.AwaitStopped()
}
5 changes: 1 addition & 4 deletions cmd/cli/flavor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,10 +31,7 @@ func flavorPluginCommand(plugins func() discovery.Plugins) *cobra.Command {
return err
}

flavorPlugin, err = flavor_plugin.NewClient(endpoint.Protocol, endpoint.Address)
if err != nil {
return err
}
flavorPlugin = flavor_plugin.NewClient(endpoint.Address)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why is the protocol being removed? We need to support named pipes per #285 and this change here will have to be undone.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm uneasy about half-supported features. Assuming that #285 is O(months) out before we break ground, i find the added clarity of support and simplicity welcome in the meantime.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That said, i don't feel strongly. So if you feel strongly about keeping the protocol plumbing on this end, i can revert.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's just get this in. Some of this may have to be reworked for Windows support but for now it's fine.


return nil
},
Expand Down
5 changes: 1 addition & 4 deletions cmd/cli/group.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,10 +35,7 @@ func groupPluginCommand(plugins func() discovery.Plugins) *cobra.Command {
return err
}

groupPlugin, err = group_plugin.NewClient(endpoint.Protocol, endpoint.Address)
if err != nil {
return err
}
groupPlugin = group_plugin.NewClient(endpoint.Address)

return nil
},
Expand Down
5 changes: 1 addition & 4 deletions cmd/cli/instance.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,10 +30,7 @@ func instancePluginCommand(plugins func() discovery.Plugins) *cobra.Command {
return err
}

instancePlugin, err = instance_plugin.NewClient(endpoint.Protocol, endpoint.Address)
if err != nil {
return err
}
instancePlugin = instance_plugin.NewClient(endpoint.Address)

return nil
},
Expand Down
4 changes: 2 additions & 2 deletions cmd/group/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,15 +40,15 @@ func main() {
if err != nil {
return nil, err
}
return instance_client.NewClient(endpoint.Protocol, endpoint.Address)
return instance_client.NewClient(endpoint.Address), nil
}

flavorPluginLookup := func(n string) (flavor.Plugin, error) {
endpoint, err := plugins.Find(n)
if err != nil {
return nil, err
}
return flavor_client.NewClient(endpoint.Protocol, endpoint.Address)
return flavor_client.NewClient(endpoint.Address), nil
}

cli.RunPlugin(name, group_server.PluginServer(
Expand Down
17 changes: 4 additions & 13 deletions cmd/manager/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@ import (
"github.com/docker/infrakit/discovery"
"github.com/docker/infrakit/leader"
"github.com/docker/infrakit/manager"
"github.com/docker/infrakit/rpc"
group_rpc "github.com/docker/infrakit/rpc/group"
"github.com/docker/infrakit/store"
"github.com/spf13/cobra"
Expand Down Expand Up @@ -52,28 +51,20 @@ func runMain(backend *backend) error {

log.Infoln("Starting up manager:", backend)

manager, err := manager.NewManager(backend.plugins,
mgr, err := manager.NewManager(backend.plugins,
backend.leader, backend.snapshot, backend.pluginName)
if err != nil {
return err
}

_, err = manager.Start()
_, err = mgr.Start()
if err != nil {
return err
}

_, stopped, err := rpc.StartPluginAtPath(
filepath.Join(discovery.Dir(), backend.id),
group_rpc.PluginServer(manager),
)
if err != nil {
return err
}

<-stopped // block until done
cli.RunPlugin(backend.id, group_rpc.PluginServer(mgr))

manager.Stop()
mgr.Stop()
log.Infoln("Manager stopped")

return err
Expand Down
20 changes: 9 additions & 11 deletions discovery/dir_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,8 @@ import (
"testing"
"time"

server "github.com/docker/infrakit/rpc"
rpc "github.com/docker/infrakit/rpc/instance"
"github.com/docker/infrakit/rpc/server"
"github.com/stretchr/testify/require"
)

Expand All @@ -31,17 +31,15 @@ func TestDirDiscovery(t *testing.T) {

name1 := "server1"
path1 := filepath.Join(dir, name1)
stop1, errors1, err1 := server.StartPluginAtPath(path1, rpc.PluginServer(nil))
require.NoError(t, err1)
require.NotNil(t, stop1)
require.NotNil(t, errors1)
server1, err := server.StartPluginAtPath(path1, rpc.PluginServer(nil))
require.NoError(t, err)
require.NotNil(t, server1)

name2 := "server2"
path2 := filepath.Join(dir, name2)
stop2, errors2, err2 := server.StartPluginAtPath(path2, rpc.PluginServer(nil))
require.NoError(t, err2)
require.NotNil(t, stop2)
require.NotNil(t, errors2)
server2, err := server.StartPluginAtPath(path2, rpc.PluginServer(nil))
require.NoError(t, err)
require.NotNil(t, server2)

discover, err := newDirPluginDiscovery(dir)
require.NoError(t, err)
Expand All @@ -55,7 +53,7 @@ func TestDirDiscovery(t *testing.T) {
require.NotNil(t, p)

// Now we stop the servers
close(stop1)
server1.Stop()
blockWhileFileExists(path1)

p, err = discover.Find(name1)
Expand All @@ -65,7 +63,7 @@ func TestDirDiscovery(t *testing.T) {
require.NoError(t, err)
require.NotNil(t, p)

close(stop2)
server2.Stop()

blockWhileFileExists(path2)

Expand Down
2 changes: 1 addition & 1 deletion example/flavor/combo/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ func main() {
if err != nil {
return nil, err
}
return flavor_rpc.NewClient(endpoint.Protocol, endpoint.Address)
return flavor_rpc.NewClient(endpoint.Address), nil
}

cli.SetLogLevel(logLevel)
Expand Down
6 changes: 4 additions & 2 deletions leader/swarm/swarm_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,10 @@ func TestSwarmDetector(t *testing.T) {

ctx := context.Background()
nodeInfo := types.Info{
Swarm: swarm.Info{
NodeID: "node",
InfoBase: &types.InfoBase{
Swarm: swarm.Info{
NodeID: "node",
},
},
}
node := swarm.Node{
Expand Down
7 changes: 1 addition & 6 deletions manager/group_plugin_impl.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,8 @@ func (m *manager) proxyForGroupPlugin(name string) (group.Plugin, error) {
return nil, err
}

client, err := rpc.NewClient(endpoint.Protocol, endpoint.Address)
if err != nil {
return nil, err
}

m.backendName = name
return client, nil
return rpc.NewClient(endpoint.Address), nil
}

// This implements the Group Plugin interface to support single group-only operations
Expand Down
2 changes: 1 addition & 1 deletion manager/manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -326,7 +326,7 @@ func (m *manager) execPlugins(config GlobalSpec, work func(group.Plugin, group.S
return err
}

gp, err := rpc.NewClient(ep.Protocol, ep.Address)
gp := rpc.NewClient(ep.Address)
if err != nil {
log.Warningln("Cannot contact group", id, " at plugin", name, "endpoint=", ep.Address)
return err
Expand Down
Loading