Skip to content

Commit 7049671

Browse files
dmcgowandmcgowan
committed
Add support for proxy plugins in configuration
Signed-off-by: Derek McGowan <[email protected]>
1 parent 63522d9 commit 7049671

File tree

3 files changed

+95
-4
lines changed

3 files changed

+95
-4
lines changed

‎cmd/containerd/command/config.go

+2-1
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
package command
1818

1919
import (
20+
gocontext "context"
2021
"io"
2122
"os"
2223

@@ -48,7 +49,7 @@ var configCommand = cli.Command{
4849
config := &Config{
4950
Config: defaultConfig(),
5051
}
51-
plugins, err := server.LoadPlugins(config.Config)
52+
plugins, err := server.LoadPlugins(gocontext.Background(), config.Config)
5253
if err != nil {
5354
return err
5455
}

‎services/server/config.go

+8
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,8 @@ type Config struct {
4343
OOMScore int `toml:"oom_score"`
4444
// Cgroup specifies cgroup information for the containerd daemon process
4545
Cgroup CgroupConfig `toml:"cgroup"`
46+
// ProxyPlugins configures plugins which are communicated to over GRPC
47+
ProxyPlugins map[string]ProxyPlugin `toml:"proxy_plugins"`
4648

4749
md toml.MetaData
4850
}
@@ -75,6 +77,12 @@ type CgroupConfig struct {
7577
Path string `toml:"path"`
7678
}
7779

80+
// ProxyPlugin provides a proxy plugin configuration
81+
type ProxyPlugin struct {
82+
Type string `toml:"type"`
83+
Address string `toml:"address"`
84+
}
85+
7886
// Decode unmarshals a plugin specific configuration by plugin id
7987
func (c *Config) Decode(id string, v interface{}) (interface{}, error) {
8088
data, ok := c.Plugins[id]

‎services/server/server.go

+85-3
Original file line numberDiff line numberDiff line change
@@ -26,19 +26,26 @@ import (
2626
"os"
2727
"path/filepath"
2828
"strings"
29+
"sync"
30+
"time"
2931

3032
"github.com/boltdb/bolt"
33+
csapi "github.com/containerd/containerd/api/services/content/v1"
34+
ssapi "github.com/containerd/containerd/api/services/snapshots/v1"
3135
"github.com/containerd/containerd/content"
3236
"github.com/containerd/containerd/content/local"
37+
csproxy "github.com/containerd/containerd/content/proxy"
38+
"github.com/containerd/containerd/defaults"
3339
"github.com/containerd/containerd/events/exchange"
3440
"github.com/containerd/containerd/log"
3541
"github.com/containerd/containerd/metadata"
42+
"github.com/containerd/containerd/pkg/dialer"
3643
"github.com/containerd/containerd/plugin"
3744
"github.com/containerd/containerd/snapshots"
45+
ssproxy "github.com/containerd/containerd/snapshots/proxy"
3846
metrics "github.com/docker/go-metrics"
3947
grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
4048
"github.com/pkg/errors"
41-
4249
"google.golang.org/grpc"
4350
)
4451

@@ -62,7 +69,7 @@ func New(ctx context.Context, config *Config) (*Server, error) {
6269
if err := apply(ctx, config); err != nil {
6370
return nil, err
6471
}
65-
plugins, err := LoadPlugins(config)
72+
plugins, err := LoadPlugins(ctx, config)
6673
if err != nil {
6774
return nil, err
6875
}
@@ -204,7 +211,7 @@ func (s *Server) Stop() {
204211

205212
// LoadPlugins loads all plugins into containerd and generates an ordered graph
206213
// of all plugins.
207-
func LoadPlugins(config *Config) ([]*plugin.Registration, error) {
214+
func LoadPlugins(ctx context.Context, config *Config) ([]*plugin.Registration, error) {
208215
// load all plugins into containerd
209216
if err := plugin.Load(filepath.Join(config.Root, "plugins")); err != nil {
210217
return nil, err
@@ -265,10 +272,85 @@ func LoadPlugins(config *Config) ([]*plugin.Registration, error) {
265272
},
266273
})
267274

275+
clients := &proxyClients{}
276+
for name, pp := range config.ProxyPlugins {
277+
var (
278+
t plugin.Type
279+
f func(*grpc.ClientConn) interface{}
280+
281+
address = pp.Address
282+
)
283+
284+
switch pp.Type {
285+
case string(plugin.SnapshotPlugin), "snapshot":
286+
t = plugin.SnapshotPlugin
287+
ssname := name
288+
f = func(conn *grpc.ClientConn) interface{} {
289+
return ssproxy.NewSnapshotter(ssapi.NewSnapshotsClient(conn), ssname)
290+
}
291+
292+
case string(plugin.ContentPlugin), "content":
293+
t = plugin.ContentPlugin
294+
f = func(conn *grpc.ClientConn) interface{} {
295+
return csproxy.NewContentStore(csapi.NewContentClient(conn))
296+
}
297+
default:
298+
log.G(ctx).WithField("type", pp.Type).Warn("unknown proxy plugin type")
299+
}
300+
301+
plugin.Register(&plugin.Registration{
302+
Type: t,
303+
ID: name,
304+
InitFn: func(ic *plugin.InitContext) (interface{}, error) {
305+
ic.Meta.Exports["address"] = address
306+
conn, err := clients.getClient(address)
307+
if err != nil {
308+
return nil, err
309+
}
310+
return f(conn), nil
311+
},
312+
})
313+
314+
}
315+
268316
// return the ordered graph for plugins
269317
return plugin.Graph(config.DisabledPlugins), nil
270318
}
271319

320+
type proxyClients struct {
321+
m sync.Mutex
322+
clients map[string]*grpc.ClientConn
323+
}
324+
325+
func (pc *proxyClients) getClient(address string) (*grpc.ClientConn, error) {
326+
pc.m.Lock()
327+
defer pc.m.Unlock()
328+
if pc.clients == nil {
329+
pc.clients = map[string]*grpc.ClientConn{}
330+
} else if c, ok := pc.clients[address]; ok {
331+
return c, nil
332+
}
333+
334+
gopts := []grpc.DialOption{
335+
grpc.WithInsecure(),
336+
grpc.WithBackoffMaxDelay(3 * time.Second),
337+
grpc.WithDialer(dialer.Dialer),
338+
339+
// TODO(stevvooe): We may need to allow configuration of this on the client.
340+
grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(defaults.DefaultMaxRecvMsgSize)),
341+
grpc.WithDefaultCallOptions(grpc.MaxCallSendMsgSize(defaults.DefaultMaxSendMsgSize)),
342+
}
343+
344+
conn, err := grpc.Dial(dialer.DialAddress(address), gopts...)
345+
if err != nil {
346+
return nil, errors.Wrapf(err, "failed to dial %q", address)
347+
}
348+
349+
pc.clients[address] = conn
350+
351+
return conn, nil
352+
}
353+
272354
func trapClosedConnErr(err error) error {
273355
if err == nil {
274356
return nil

0 commit comments

Comments
 (0)