mirror of
https://github.com/moby/moby.git
synced 2022-11-09 12:21:53 -05:00
remove pkg/discovery as it's now unused
Signed-off-by: Sebastiaan van Stijn <github@gone.nl>
This commit is contained in:
Sebastiaan van Stijn
parent
00f9b23c3a
commit
e202ac3f38
15 changed files with 0 additions and 1482 deletions
|
|
@ -1,41 +0,0 @@
|
|||
---
|
||||
page_title: Docker discovery
|
||||
page_description: discovery
|
||||
page_keywords: docker, clustering, discovery
|
||||
---
|
||||
|
||||
# Discovery
|
||||
|
||||
Docker comes with multiple Discovery backends.
|
||||
|
||||
## Backends
|
||||
|
||||
### Using etcd
|
||||
|
||||
Point your Docker Engine instances to a common etcd instance. You can specify
|
||||
the address Docker uses to advertise the node using the `--cluster-advertise`
|
||||
flag.
|
||||
|
||||
```bash
|
||||
$ dockerd -H=<node_ip:2376> --cluster-advertise=<node_ip:2376> --cluster-store etcd://<etcd_ip1>,<etcd_ip2>/<path>
|
||||
```
|
||||
|
||||
### Using consul
|
||||
|
||||
Point your Docker Engine instances to a common Consul instance. You can specify
|
||||
the address Docker uses to advertise the node using the `--cluster-advertise`
|
||||
flag.
|
||||
|
||||
```bash
|
||||
$ dockerd -H=<node_ip:2376> --cluster-advertise=<node_ip:2376> --cluster-store consul://<consul_ip>/<path>
|
||||
```
|
||||
|
||||
### Using zookeeper
|
||||
|
||||
Point your Docker Engine instances to a common Zookeeper instance. You can specify
|
||||
the address Docker uses to advertise the node using the `--cluster-advertise`
|
||||
flag.
|
||||
|
||||
```bash
|
||||
$ dockerd -H=<node_ip:2376> --cluster-advertise=<node_ip:2376> --cluster-store zk://<zk_addr1>,<zk_addr2>/<path>
|
||||
```
|
||||
|
|
@ -1,107 +0,0 @@
|
|||
package discovery // import "github.com/docker/docker/pkg/discovery"
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
var (
|
||||
// Backends is a global map of discovery backends indexed by their
|
||||
// associated scheme.
|
||||
backends = make(map[string]Backend)
|
||||
)
|
||||
|
||||
// Register makes a discovery backend available by the provided scheme.
|
||||
// If Register is called twice with the same scheme an error is returned.
|
||||
func Register(scheme string, d Backend) error {
|
||||
if _, exists := backends[scheme]; exists {
|
||||
return fmt.Errorf("scheme already registered %s", scheme)
|
||||
}
|
||||
logrus.WithField("name", scheme).Debugf("Registering discovery service")
|
||||
backends[scheme] = d
|
||||
return nil
|
||||
}
|
||||
|
||||
func parse(rawurl string) (string, string) {
|
||||
parts := strings.SplitN(rawurl, "://", 2)
|
||||
|
||||
// nodes:port,node2:port => nodes://node1:port,node2:port
|
||||
if len(parts) == 1 {
|
||||
return "nodes", parts[0]
|
||||
}
|
||||
return parts[0], parts[1]
|
||||
}
|
||||
|
||||
// ParseAdvertise parses the --cluster-advertise daemon config which accepts
|
||||
// <ip-address>:<port> or <interface-name>:<port>
|
||||
func ParseAdvertise(advertise string) (string, error) {
|
||||
var (
|
||||
iface *net.Interface
|
||||
addrs []net.Addr
|
||||
err error
|
||||
)
|
||||
|
||||
addr, port, err := net.SplitHostPort(advertise)
|
||||
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("invalid --cluster-advertise configuration: %s: %v", advertise, err)
|
||||
}
|
||||
|
||||
ip := net.ParseIP(addr)
|
||||
// If it is a valid ip-address, use it as is
|
||||
if ip != nil {
|
||||
return advertise, nil
|
||||
}
|
||||
|
||||
// If advertise is a valid interface name, get the valid IPv4 address and use it to advertise
|
||||
ifaceName := addr
|
||||
iface, err = net.InterfaceByName(ifaceName)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("invalid cluster advertise IP address or interface name (%s) : %v", advertise, err)
|
||||
}
|
||||
|
||||
addrs, err = iface.Addrs()
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("unable to get advertise IP address from interface (%s) : %v", advertise, err)
|
||||
}
|
||||
|
||||
if len(addrs) == 0 {
|
||||
return "", fmt.Errorf("no available advertise IP address in interface (%s)", advertise)
|
||||
}
|
||||
|
||||
addr = ""
|
||||
for _, a := range addrs {
|
||||
ip, _, err := net.ParseCIDR(a.String())
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("error deriving advertise ip-address in interface (%s) : %v", advertise, err)
|
||||
}
|
||||
if ip.To4() == nil || ip.IsLoopback() {
|
||||
continue
|
||||
}
|
||||
addr = ip.String()
|
||||
break
|
||||
}
|
||||
if addr == "" {
|
||||
return "", fmt.Errorf("could not find a valid ip-address in interface %s", advertise)
|
||||
}
|
||||
|
||||
addr = net.JoinHostPort(addr, port)
|
||||
return addr, nil
|
||||
}
|
||||
|
||||
// New returns a new Discovery given a URL, heartbeat and ttl settings.
|
||||
// Returns an error if the URL scheme is not supported.
|
||||
func New(rawurl string, heartbeat time.Duration, ttl time.Duration, clusterOpts map[string]string) (Backend, error) {
|
||||
scheme, uri := parse(rawurl)
|
||||
if backend, exists := backends[scheme]; exists {
|
||||
logrus.WithFields(logrus.Fields{"name": scheme, "uri": uri}).Debugf("Initializing discovery service")
|
||||
err := backend.Initialize(uri, heartbeat, ttl, clusterOpts)
|
||||
return backend, err
|
||||
}
|
||||
|
||||
return nil, ErrNotSupported
|
||||
}
|
||||
|
|
@ -1,35 +0,0 @@
|
|||
package discovery // import "github.com/docker/docker/pkg/discovery"
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"time"
|
||||
)
|
||||
|
||||
var (
|
||||
// ErrNotSupported is returned when a discovery service is not supported.
|
||||
ErrNotSupported = errors.New("discovery service not supported")
|
||||
|
||||
// ErrNotImplemented is returned when discovery feature is not implemented
|
||||
// by discovery backend.
|
||||
ErrNotImplemented = errors.New("not implemented in this discovery service")
|
||||
)
|
||||
|
||||
// Watcher provides watching over a cluster for nodes joining and leaving.
|
||||
type Watcher interface {
|
||||
// Watch the discovery for entry changes.
|
||||
// Returns a channel that will receive changes or an error.
|
||||
// Providing a non-nil stopCh can be used to stop watching.
|
||||
Watch(stopCh <-chan struct{}) (<-chan Entries, <-chan error)
|
||||
}
|
||||
|
||||
// Backend is implemented by discovery backends which manage cluster entries.
|
||||
type Backend interface {
|
||||
// Watcher must be provided by every backend.
|
||||
Watcher
|
||||
|
||||
// Initialize the discovery with URIs, a heartbeat, a ttl and optional settings.
|
||||
Initialize(string, time.Duration, time.Duration, map[string]string) error
|
||||
|
||||
// Register to the discovery.
|
||||
Register(string) error
|
||||
}
|
||||
|
|
@ -1,135 +0,0 @@
|
|||
package discovery // import "github.com/docker/docker/pkg/discovery"
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/docker/docker/internal/test/suite"
|
||||
"gotest.tools/v3/assert"
|
||||
)
|
||||
|
||||
// Hook up gocheck into the "go test" runner.
|
||||
func Test(t *testing.T) {
|
||||
suite.Run(t, &DiscoverySuite{})
|
||||
}
|
||||
|
||||
type DiscoverySuite struct{}
|
||||
|
||||
func (s *DiscoverySuite) TestNewEntry(c *testing.T) {
|
||||
entry, err := NewEntry("127.0.0.1:2375")
|
||||
assert.Assert(c, err == nil)
|
||||
assert.Equal(c, entry.Equals(&Entry{Host: "127.0.0.1", Port: "2375"}), true)
|
||||
assert.Equal(c, entry.String(), "127.0.0.1:2375")
|
||||
|
||||
entry, err = NewEntry("[2001:db8:0:f101::2]:2375")
|
||||
assert.Assert(c, err == nil)
|
||||
assert.Equal(c, entry.Equals(&Entry{Host: "2001:db8:0:f101::2", Port: "2375"}), true)
|
||||
assert.Equal(c, entry.String(), "[2001:db8:0:f101::2]:2375")
|
||||
|
||||
_, err = NewEntry("127.0.0.1")
|
||||
assert.Assert(c, err != nil)
|
||||
}
|
||||
|
||||
func (s *DiscoverySuite) TestParse(c *testing.T) {
|
||||
scheme, uri := parse("127.0.0.1:2375")
|
||||
assert.Equal(c, scheme, "nodes")
|
||||
assert.Equal(c, uri, "127.0.0.1:2375")
|
||||
|
||||
scheme, uri = parse("localhost:2375")
|
||||
assert.Equal(c, scheme, "nodes")
|
||||
assert.Equal(c, uri, "localhost:2375")
|
||||
|
||||
scheme, uri = parse("scheme://127.0.0.1:2375")
|
||||
assert.Equal(c, scheme, "scheme")
|
||||
assert.Equal(c, uri, "127.0.0.1:2375")
|
||||
|
||||
scheme, uri = parse("scheme://localhost:2375")
|
||||
assert.Equal(c, scheme, "scheme")
|
||||
assert.Equal(c, uri, "localhost:2375")
|
||||
|
||||
scheme, uri = parse("")
|
||||
assert.Equal(c, scheme, "nodes")
|
||||
assert.Equal(c, uri, "")
|
||||
}
|
||||
|
||||
func (s *DiscoverySuite) TestCreateEntries(c *testing.T) {
|
||||
entries, err := CreateEntries(nil)
|
||||
assert.DeepEqual(c, entries, Entries{})
|
||||
assert.Assert(c, err == nil)
|
||||
|
||||
entries, err = CreateEntries([]string{"127.0.0.1:2375", "127.0.0.2:2375", "[2001:db8:0:f101::2]:2375", ""})
|
||||
assert.Assert(c, err == nil)
|
||||
expected := Entries{
|
||||
&Entry{Host: "127.0.0.1", Port: "2375"},
|
||||
&Entry{Host: "127.0.0.2", Port: "2375"},
|
||||
&Entry{Host: "2001:db8:0:f101::2", Port: "2375"},
|
||||
}
|
||||
assert.Equal(c, entries.Equals(expected), true)
|
||||
|
||||
_, err = CreateEntries([]string{"127.0.0.1", "127.0.0.2"})
|
||||
assert.Assert(c, err != nil)
|
||||
}
|
||||
|
||||
func (s *DiscoverySuite) TestContainsEntry(c *testing.T) {
|
||||
entries, err := CreateEntries([]string{"127.0.0.1:2375", "127.0.0.2:2375", ""})
|
||||
assert.Assert(c, err == nil)
|
||||
assert.Equal(c, entries.Contains(&Entry{Host: "127.0.0.1", Port: "2375"}), true)
|
||||
assert.Equal(c, entries.Contains(&Entry{Host: "127.0.0.3", Port: "2375"}), false)
|
||||
}
|
||||
|
||||
func (s *DiscoverySuite) TestEntriesEquality(c *testing.T) {
|
||||
entries := Entries{
|
||||
&Entry{Host: "127.0.0.1", Port: "2375"},
|
||||
&Entry{Host: "127.0.0.2", Port: "2375"},
|
||||
}
|
||||
|
||||
// Same
|
||||
assert.Assert(c, entries.Equals(Entries{
|
||||
&Entry{Host: "127.0.0.1", Port: "2375"},
|
||||
&Entry{Host: "127.0.0.2", Port: "2375"},
|
||||
}))
|
||||
|
||||
// Different size
|
||||
assert.Assert(c, !entries.Equals(Entries{
|
||||
&Entry{Host: "127.0.0.1", Port: "2375"},
|
||||
&Entry{Host: "127.0.0.2", Port: "2375"},
|
||||
&Entry{Host: "127.0.0.3", Port: "2375"},
|
||||
}))
|
||||
|
||||
// Different content
|
||||
assert.Assert(c, !entries.Equals(Entries{
|
||||
&Entry{Host: "127.0.0.1", Port: "2375"},
|
||||
&Entry{Host: "127.0.0.42", Port: "2375"},
|
||||
}))
|
||||
|
||||
}
|
||||
|
||||
func (s *DiscoverySuite) TestEntriesDiff(c *testing.T) {
|
||||
entry1 := &Entry{Host: "1.1.1.1", Port: "1111"}
|
||||
entry2 := &Entry{Host: "2.2.2.2", Port: "2222"}
|
||||
entry3 := &Entry{Host: "3.3.3.3", Port: "3333"}
|
||||
entries := Entries{entry1, entry2}
|
||||
|
||||
// No diff
|
||||
added, removed := entries.Diff(Entries{entry2, entry1})
|
||||
assert.Equal(c, len(added), 0)
|
||||
assert.Equal(c, len(removed), 0)
|
||||
|
||||
// Add
|
||||
added, removed = entries.Diff(Entries{entry2, entry3, entry1})
|
||||
assert.Equal(c, len(added), 1)
|
||||
assert.Equal(c, added.Contains(entry3), true)
|
||||
assert.Equal(c, len(removed), 0)
|
||||
|
||||
// Remove
|
||||
added, removed = entries.Diff(Entries{entry2})
|
||||
assert.Equal(c, len(added), 0)
|
||||
assert.Equal(c, len(removed), 1)
|
||||
assert.Equal(c, removed.Contains(entry1), true)
|
||||
|
||||
// Add and remove
|
||||
added, removed = entries.Diff(Entries{entry1, entry3})
|
||||
assert.Equal(c, len(added), 1)
|
||||
assert.Equal(c, added.Contains(entry3), true)
|
||||
assert.Equal(c, len(removed), 1)
|
||||
assert.Equal(c, removed.Contains(entry2), true)
|
||||
}
|
||||
|
|
@ -1,94 +0,0 @@
|
|||
package discovery // import "github.com/docker/docker/pkg/discovery"
|
||||
|
||||
import "net"
|
||||
|
||||
// NewEntry creates a new entry.
|
||||
func NewEntry(url string) (*Entry, error) {
|
||||
host, port, err := net.SplitHostPort(url)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &Entry{host, port}, nil
|
||||
}
|
||||
|
||||
// An Entry represents a host.
|
||||
type Entry struct {
|
||||
Host string
|
||||
Port string
|
||||
}
|
||||
|
||||
// Equals returns true if cmp contains the same data.
|
||||
func (e *Entry) Equals(cmp *Entry) bool {
|
||||
return e.Host == cmp.Host && e.Port == cmp.Port
|
||||
}
|
||||
|
||||
// String returns the string form of an entry.
|
||||
func (e *Entry) String() string {
|
||||
return net.JoinHostPort(e.Host, e.Port)
|
||||
}
|
||||
|
||||
// Entries is a list of *Entry with some helpers.
|
||||
type Entries []*Entry
|
||||
|
||||
// Equals returns true if cmp contains the same data.
|
||||
func (e Entries) Equals(cmp Entries) bool {
|
||||
// Check if the file has really changed.
|
||||
if len(e) != len(cmp) {
|
||||
return false
|
||||
}
|
||||
for i := range e {
|
||||
if !e[i].Equals(cmp[i]) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// Contains returns true if the Entries contain a given Entry.
|
||||
func (e Entries) Contains(entry *Entry) bool {
|
||||
for _, curr := range e {
|
||||
if curr.Equals(entry) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// Diff compares two entries and returns the added and removed entries.
|
||||
func (e Entries) Diff(cmp Entries) (Entries, Entries) {
|
||||
added := Entries{}
|
||||
for _, entry := range cmp {
|
||||
if !e.Contains(entry) {
|
||||
added = append(added, entry)
|
||||
}
|
||||
}
|
||||
|
||||
removed := Entries{}
|
||||
for _, entry := range e {
|
||||
if !cmp.Contains(entry) {
|
||||
removed = append(removed, entry)
|
||||
}
|
||||
}
|
||||
|
||||
return added, removed
|
||||
}
|
||||
|
||||
// CreateEntries returns an array of entries based on the given addresses.
|
||||
func CreateEntries(addrs []string) (Entries, error) {
|
||||
entries := Entries{}
|
||||
if addrs == nil {
|
||||
return entries, nil
|
||||
}
|
||||
|
||||
for _, addr := range addrs {
|
||||
if len(addr) == 0 {
|
||||
continue
|
||||
}
|
||||
entry, err := NewEntry(addr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
entries = append(entries, entry)
|
||||
}
|
||||
return entries, nil
|
||||
}
|
||||
|
|
@ -1,107 +0,0 @@
|
|||
package file // import "github.com/docker/docker/pkg/discovery/file"
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/docker/docker/pkg/discovery"
|
||||
)
|
||||
|
||||
// Discovery is exported
|
||||
type Discovery struct {
|
||||
heartbeat time.Duration
|
||||
path string
|
||||
}
|
||||
|
||||
func init() {
|
||||
Init()
|
||||
}
|
||||
|
||||
// Init is exported
|
||||
func Init() {
|
||||
discovery.Register("file", &Discovery{})
|
||||
}
|
||||
|
||||
// Initialize is exported
|
||||
func (s *Discovery) Initialize(path string, heartbeat time.Duration, ttl time.Duration, _ map[string]string) error {
|
||||
s.path = path
|
||||
s.heartbeat = heartbeat
|
||||
return nil
|
||||
}
|
||||
|
||||
func parseFileContent(content []byte) []string {
|
||||
var result []string
|
||||
for _, line := range strings.Split(strings.TrimSpace(string(content)), "\n") {
|
||||
line = strings.TrimSpace(line)
|
||||
// Ignoring line starts with #
|
||||
if strings.HasPrefix(line, "#") {
|
||||
continue
|
||||
}
|
||||
// Inlined # comment also ignored.
|
||||
if strings.Contains(line, "#") {
|
||||
line = line[0:strings.Index(line, "#")]
|
||||
// Trim additional spaces caused by above stripping.
|
||||
line = strings.TrimSpace(line)
|
||||
}
|
||||
result = append(result, discovery.Generate(line)...)
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func (s *Discovery) fetch() (discovery.Entries, error) {
|
||||
fileContent, err := os.ReadFile(s.path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to read '%s': %v", s.path, err)
|
||||
}
|
||||
return discovery.CreateEntries(parseFileContent(fileContent))
|
||||
}
|
||||
|
||||
// Watch is exported
|
||||
func (s *Discovery) Watch(stopCh <-chan struct{}) (<-chan discovery.Entries, <-chan error) {
|
||||
ch := make(chan discovery.Entries, 1)
|
||||
errCh := make(chan error, 1)
|
||||
ticker := time.NewTicker(s.heartbeat)
|
||||
|
||||
go func() {
|
||||
defer close(errCh)
|
||||
defer close(ch)
|
||||
|
||||
// Send the initial entries if available.
|
||||
currentEntries, err := s.fetch()
|
||||
if err != nil {
|
||||
errCh <- err
|
||||
} else {
|
||||
ch <- currentEntries
|
||||
}
|
||||
|
||||
// Periodically send updates.
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
newEntries, err := s.fetch()
|
||||
if err != nil {
|
||||
errCh <- err
|
||||
continue
|
||||
}
|
||||
|
||||
// Check if the file has really changed.
|
||||
if !newEntries.Equals(currentEntries) {
|
||||
ch <- newEntries
|
||||
}
|
||||
currentEntries = newEntries
|
||||
case <-stopCh:
|
||||
ticker.Stop()
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
return ch, errCh
|
||||
}
|
||||
|
||||
// Register is exported
|
||||
func (s *Discovery) Register(addr string) error {
|
||||
return discovery.ErrNotImplemented
|
||||
}
|
||||
|
|
@ -1,113 +0,0 @@
|
|||
package file // import "github.com/docker/docker/pkg/discovery/file"
|
||||
|
||||
import (
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/docker/docker/internal/test/suite"
|
||||
"github.com/docker/docker/pkg/discovery"
|
||||
"gotest.tools/v3/assert"
|
||||
)
|
||||
|
||||
// Hook up gocheck into the "go test" runner.
|
||||
func Test(t *testing.T) {
|
||||
suite.Run(t, &DiscoverySuite{})
|
||||
}
|
||||
|
||||
type DiscoverySuite struct{}
|
||||
|
||||
func (s *DiscoverySuite) TestInitialize(c *testing.T) {
|
||||
d := &Discovery{}
|
||||
d.Initialize("/path/to/file", 1000, 0, nil)
|
||||
assert.Equal(c, d.path, "/path/to/file")
|
||||
}
|
||||
|
||||
func (s *DiscoverySuite) TestNew(c *testing.T) {
|
||||
d, err := discovery.New("file:///path/to/file", 0, 0, nil)
|
||||
assert.Assert(c, err == nil)
|
||||
assert.Equal(c, d.(*Discovery).path, "/path/to/file")
|
||||
}
|
||||
|
||||
func (s *DiscoverySuite) TestContent(c *testing.T) {
|
||||
data := `
|
||||
1.1.1.[1:2]:1111
|
||||
2.2.2.[2:4]:2222
|
||||
`
|
||||
ips := parseFileContent([]byte(data))
|
||||
assert.Equal(c, len(ips), 5)
|
||||
assert.Equal(c, ips[0], "1.1.1.1:1111")
|
||||
assert.Equal(c, ips[1], "1.1.1.2:1111")
|
||||
assert.Equal(c, ips[2], "2.2.2.2:2222")
|
||||
assert.Equal(c, ips[3], "2.2.2.3:2222")
|
||||
assert.Equal(c, ips[4], "2.2.2.4:2222")
|
||||
}
|
||||
|
||||
func (s *DiscoverySuite) TestRegister(c *testing.T) {
|
||||
discovery := &Discovery{path: "/path/to/file"}
|
||||
assert.Assert(c, discovery.Register("0.0.0.0") != nil)
|
||||
}
|
||||
|
||||
func (s *DiscoverySuite) TestParsingContentsWithComments(c *testing.T) {
|
||||
data := `
|
||||
### test ###
|
||||
1.1.1.1:1111 # inline comment
|
||||
# 2.2.2.2:2222
|
||||
### empty line with comment
|
||||
3.3.3.3:3333
|
||||
### test ###
|
||||
`
|
||||
ips := parseFileContent([]byte(data))
|
||||
assert.Equal(c, len(ips), 2)
|
||||
assert.Equal(c, "1.1.1.1:1111", ips[0])
|
||||
assert.Equal(c, "3.3.3.3:3333", ips[1])
|
||||
}
|
||||
|
||||
func (s *DiscoverySuite) TestWatch(c *testing.T) {
|
||||
data := `
|
||||
1.1.1.1:1111
|
||||
2.2.2.2:2222
|
||||
`
|
||||
expected := discovery.Entries{
|
||||
&discovery.Entry{Host: "1.1.1.1", Port: "1111"},
|
||||
&discovery.Entry{Host: "2.2.2.2", Port: "2222"},
|
||||
}
|
||||
|
||||
// Create a temporary file and remove it.
|
||||
tmp, err := os.CreateTemp(os.TempDir(), "discovery-file-test")
|
||||
assert.Assert(c, err == nil)
|
||||
assert.Assert(c, tmp.Close() == nil)
|
||||
assert.Assert(c, os.Remove(tmp.Name()) == nil)
|
||||
|
||||
// Set up file discovery.
|
||||
d := &Discovery{}
|
||||
d.Initialize(tmp.Name(), 1000, 0, nil)
|
||||
stopCh := make(chan struct{})
|
||||
ch, errCh := d.Watch(stopCh)
|
||||
|
||||
// Make sure it fires errors since the file doesn't exist.
|
||||
assert.Assert(c, <-errCh != nil)
|
||||
// We have to drain the error channel otherwise Watch will get stuck.
|
||||
go func() {
|
||||
for range errCh {
|
||||
}
|
||||
}()
|
||||
|
||||
// Write the file and make sure we get the expected value back.
|
||||
assert.Assert(c, os.WriteFile(tmp.Name(), []byte(data), 0600) == nil)
|
||||
assert.DeepEqual(c, <-ch, expected)
|
||||
|
||||
// Add a new entry and look it up.
|
||||
expected = append(expected, &discovery.Entry{Host: "3.3.3.3", Port: "3333"})
|
||||
f, err := os.OpenFile(tmp.Name(), os.O_APPEND|os.O_WRONLY, 0600)
|
||||
assert.Assert(c, err == nil)
|
||||
assert.Assert(c, f != nil)
|
||||
_, err = f.WriteString("\n3.3.3.3:3333\n")
|
||||
assert.Assert(c, err == nil)
|
||||
f.Close()
|
||||
assert.DeepEqual(c, <-ch, expected)
|
||||
|
||||
// Stop and make sure it closes all channels.
|
||||
close(stopCh)
|
||||
assert.Assert(c, <-ch == nil)
|
||||
assert.Assert(c, <-errCh == nil)
|
||||
}
|
||||
|
|
@ -1,35 +0,0 @@
|
|||
package discovery // import "github.com/docker/docker/pkg/discovery"
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"regexp"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
// Generate takes care of IP generation
|
||||
func Generate(pattern string) []string {
|
||||
re, _ := regexp.Compile(`\[(.+):(.+)\]`)
|
||||
submatch := re.FindStringSubmatch(pattern)
|
||||
if submatch == nil {
|
||||
return []string{pattern}
|
||||
}
|
||||
|
||||
from, err := strconv.Atoi(submatch[1])
|
||||
if err != nil {
|
||||
return []string{pattern}
|
||||
}
|
||||
to, err := strconv.Atoi(submatch[2])
|
||||
if err != nil {
|
||||
return []string{pattern}
|
||||
}
|
||||
|
||||
template := re.ReplaceAllString(pattern, "%d")
|
||||
|
||||
var result []string
|
||||
for val := from; val <= to; val++ {
|
||||
entry := fmt.Sprintf(template, val)
|
||||
result = append(result, entry)
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
|
@ -1,54 +0,0 @@
|
|||
package discovery // import "github.com/docker/docker/pkg/discovery"
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"gotest.tools/v3/assert"
|
||||
)
|
||||
|
||||
func (s *DiscoverySuite) TestGeneratorNotGenerate(c *testing.T) {
|
||||
ips := Generate("127.0.0.1")
|
||||
assert.Equal(c, len(ips), 1)
|
||||
assert.Equal(c, ips[0], "127.0.0.1")
|
||||
}
|
||||
|
||||
func (s *DiscoverySuite) TestGeneratorWithPortNotGenerate(c *testing.T) {
|
||||
ips := Generate("127.0.0.1:8080")
|
||||
assert.Equal(c, len(ips), 1)
|
||||
assert.Equal(c, ips[0], "127.0.0.1:8080")
|
||||
}
|
||||
|
||||
func (s *DiscoverySuite) TestGeneratorMatchFailedNotGenerate(c *testing.T) {
|
||||
ips := Generate("127.0.0.[1]")
|
||||
assert.Equal(c, len(ips), 1)
|
||||
assert.Equal(c, ips[0], "127.0.0.[1]")
|
||||
}
|
||||
|
||||
func (s *DiscoverySuite) TestGeneratorWithPort(c *testing.T) {
|
||||
ips := Generate("127.0.0.[1:11]:2375")
|
||||
assert.Equal(c, len(ips), 11)
|
||||
assert.Equal(c, ips[0], "127.0.0.1:2375")
|
||||
assert.Equal(c, ips[1], "127.0.0.2:2375")
|
||||
assert.Equal(c, ips[2], "127.0.0.3:2375")
|
||||
assert.Equal(c, ips[3], "127.0.0.4:2375")
|
||||
assert.Equal(c, ips[4], "127.0.0.5:2375")
|
||||
assert.Equal(c, ips[5], "127.0.0.6:2375")
|
||||
assert.Equal(c, ips[6], "127.0.0.7:2375")
|
||||
assert.Equal(c, ips[7], "127.0.0.8:2375")
|
||||
assert.Equal(c, ips[8], "127.0.0.9:2375")
|
||||
assert.Equal(c, ips[9], "127.0.0.10:2375")
|
||||
assert.Equal(c, ips[10], "127.0.0.11:2375")
|
||||
}
|
||||
|
||||
func (s *DiscoverySuite) TestGenerateWithMalformedInputAtRangeStart(c *testing.T) {
|
||||
malformedInput := "127.0.0.[x:11]:2375"
|
||||
ips := Generate(malformedInput)
|
||||
assert.Equal(c, len(ips), 1)
|
||||
assert.Equal(c, ips[0], malformedInput)
|
||||
}
|
||||
|
||||
func (s *DiscoverySuite) TestGenerateWithMalformedInputAtRangeEnd(c *testing.T) {
|
||||
malformedInput := "127.0.0.[1:x]:2375"
|
||||
ips := Generate(malformedInput)
|
||||
assert.Equal(c, len(ips), 1)
|
||||
assert.Equal(c, ips[0], malformedInput)
|
||||
}
|
||||
|
|
@ -1,192 +0,0 @@
|
|||
package kv // import "github.com/docker/docker/pkg/discovery/kv"
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"path"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/docker/docker/pkg/discovery"
|
||||
"github.com/docker/go-connections/tlsconfig"
|
||||
"github.com/docker/libkv"
|
||||
"github.com/docker/libkv/store"
|
||||
"github.com/docker/libkv/store/consul"
|
||||
"github.com/docker/libkv/store/etcd"
|
||||
"github.com/docker/libkv/store/zookeeper"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultDiscoveryPath = "docker/nodes"
|
||||
)
|
||||
|
||||
// Discovery is exported
|
||||
type Discovery struct {
|
||||
backend store.Backend
|
||||
store store.Store
|
||||
heartbeat time.Duration
|
||||
ttl time.Duration
|
||||
prefix string
|
||||
path string
|
||||
}
|
||||
|
||||
func init() {
|
||||
Init()
|
||||
}
|
||||
|
||||
// Init is exported
|
||||
func Init() {
|
||||
// Register to libkv
|
||||
zookeeper.Register()
|
||||
consul.Register()
|
||||
etcd.Register()
|
||||
|
||||
// Register to internal discovery service
|
||||
discovery.Register("zk", &Discovery{backend: store.ZK})
|
||||
discovery.Register("consul", &Discovery{backend: store.CONSUL})
|
||||
discovery.Register("etcd", &Discovery{backend: store.ETCD})
|
||||
}
|
||||
|
||||
// Initialize is exported
|
||||
func (s *Discovery) Initialize(uris string, heartbeat time.Duration, ttl time.Duration, clusterOpts map[string]string) error {
|
||||
var (
|
||||
parts = strings.SplitN(uris, "/", 2)
|
||||
addrs = strings.Split(parts[0], ",")
|
||||
err error
|
||||
)
|
||||
|
||||
// A custom prefix to the path can be optionally used.
|
||||
if len(parts) == 2 {
|
||||
s.prefix = parts[1]
|
||||
}
|
||||
|
||||
s.heartbeat = heartbeat
|
||||
s.ttl = ttl
|
||||
|
||||
// Use a custom path if specified in discovery options
|
||||
dpath := defaultDiscoveryPath
|
||||
if clusterOpts["kv.path"] != "" {
|
||||
dpath = clusterOpts["kv.path"]
|
||||
}
|
||||
|
||||
s.path = path.Join(s.prefix, dpath)
|
||||
|
||||
var config *store.Config
|
||||
if clusterOpts["kv.cacertfile"] != "" && clusterOpts["kv.certfile"] != "" && clusterOpts["kv.keyfile"] != "" {
|
||||
logrus.Info("Initializing discovery with TLS")
|
||||
tlsConfig, err := tlsconfig.Client(tlsconfig.Options{
|
||||
CAFile: clusterOpts["kv.cacertfile"],
|
||||
CertFile: clusterOpts["kv.certfile"],
|
||||
KeyFile: clusterOpts["kv.keyfile"],
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
config = &store.Config{
|
||||
// Set ClientTLS to trigger https (bug in libkv/etcd)
|
||||
ClientTLS: &store.ClientTLSConfig{
|
||||
CACertFile: clusterOpts["kv.cacertfile"],
|
||||
CertFile: clusterOpts["kv.certfile"],
|
||||
KeyFile: clusterOpts["kv.keyfile"],
|
||||
},
|
||||
// The actual TLS config that will be used
|
||||
TLS: tlsConfig,
|
||||
}
|
||||
} else {
|
||||
logrus.Info("Initializing discovery without TLS")
|
||||
}
|
||||
|
||||
// Creates a new store, will ignore options given
|
||||
// if not supported by the chosen store
|
||||
s.store, err = libkv.NewStore(s.backend, addrs, config)
|
||||
return err
|
||||
}
|
||||
|
||||
// Watch the store until either there's a store error or we receive a stop request.
|
||||
// Returns false if we shouldn't attempt watching the store anymore (stop request received).
|
||||
func (s *Discovery) watchOnce(stopCh <-chan struct{}, watchCh <-chan []*store.KVPair, discoveryCh chan discovery.Entries, errCh chan error) bool {
|
||||
for {
|
||||
select {
|
||||
case pairs := <-watchCh:
|
||||
if pairs == nil {
|
||||
return true
|
||||
}
|
||||
|
||||
logrus.WithField("discovery", s.backend).Debugf("Watch triggered with %d nodes", len(pairs))
|
||||
|
||||
// Convert `KVPair` into `discovery.Entry`.
|
||||
addrs := make([]string, len(pairs))
|
||||
for _, pair := range pairs {
|
||||
addrs = append(addrs, string(pair.Value))
|
||||
}
|
||||
|
||||
entries, err := discovery.CreateEntries(addrs)
|
||||
if err != nil {
|
||||
errCh <- err
|
||||
} else {
|
||||
discoveryCh <- entries
|
||||
}
|
||||
case <-stopCh:
|
||||
// We were requested to stop watching.
|
||||
return false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Watch is exported
|
||||
func (s *Discovery) Watch(stopCh <-chan struct{}) (<-chan discovery.Entries, <-chan error) {
|
||||
ch := make(chan discovery.Entries)
|
||||
errCh := make(chan error)
|
||||
|
||||
go func() {
|
||||
defer close(ch)
|
||||
defer close(errCh)
|
||||
|
||||
// Forever: Create a store watch, watch until we get an error and then try again.
|
||||
// Will only stop if we receive a stopCh request.
|
||||
for {
|
||||
// Create the path to watch if it does not exist yet
|
||||
exists, err := s.store.Exists(s.path)
|
||||
if err != nil {
|
||||
errCh <- err
|
||||
}
|
||||
if !exists {
|
||||
if err := s.store.Put(s.path, []byte(""), &store.WriteOptions{IsDir: true}); err != nil {
|
||||
errCh <- err
|
||||
}
|
||||
}
|
||||
|
||||
// Set up a watch.
|
||||
watchCh, err := s.store.WatchTree(s.path, stopCh)
|
||||
if err != nil {
|
||||
errCh <- err
|
||||
} else {
|
||||
if !s.watchOnce(stopCh, watchCh, ch, errCh) {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// If we get here it means the store watch channel was closed. This
|
||||
// is unexpected so let's retry later.
|
||||
errCh <- fmt.Errorf("Unexpected watch error")
|
||||
time.Sleep(s.heartbeat)
|
||||
}
|
||||
}()
|
||||
return ch, errCh
|
||||
}
|
||||
|
||||
// Register is exported
|
||||
func (s *Discovery) Register(addr string) error {
|
||||
opts := &store.WriteOptions{TTL: s.ttl}
|
||||
return s.store.Put(path.Join(s.path, addr), []byte(addr), opts)
|
||||
}
|
||||
|
||||
// Store returns the underlying store used by KV discovery.
|
||||
func (s *Discovery) Store() store.Store {
|
||||
return s.store
|
||||
}
|
||||
|
||||
// Prefix returns the store prefix
|
||||
func (s *Discovery) Prefix() string {
|
||||
return s.prefix
|
||||
}
|
||||
|
|
@ -1,322 +0,0 @@
|
|||
package kv // import "github.com/docker/docker/pkg/discovery/kv"
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"os"
|
||||
"path"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/docker/docker/internal/test/suite"
|
||||
"github.com/docker/docker/pkg/discovery"
|
||||
"github.com/docker/libkv"
|
||||
"github.com/docker/libkv/store"
|
||||
"gotest.tools/v3/assert"
|
||||
)
|
||||
|
||||
// Hook up gocheck into the "go test" runner.
|
||||
func Test(t *testing.T) {
|
||||
suite.Run(t, &DiscoverySuite{})
|
||||
}
|
||||
|
||||
type DiscoverySuite struct{}
|
||||
|
||||
func (ds *DiscoverySuite) TestInitialize(c *testing.T) {
|
||||
storeMock := &FakeStore{
|
||||
Endpoints: []string{"127.0.0.1"},
|
||||
}
|
||||
d := &Discovery{backend: store.CONSUL}
|
||||
d.Initialize("127.0.0.1", 0, 0, nil)
|
||||
d.store = storeMock
|
||||
|
||||
s := d.store.(*FakeStore)
|
||||
assert.Equal(c, len(s.Endpoints), 1)
|
||||
assert.Equal(c, s.Endpoints[0], "127.0.0.1")
|
||||
assert.Equal(c, d.path, defaultDiscoveryPath)
|
||||
|
||||
storeMock = &FakeStore{
|
||||
Endpoints: []string{"127.0.0.1:1234"},
|
||||
}
|
||||
d = &Discovery{backend: store.CONSUL}
|
||||
d.Initialize("127.0.0.1:1234/path", 0, 0, nil)
|
||||
d.store = storeMock
|
||||
|
||||
s = d.store.(*FakeStore)
|
||||
assert.Equal(c, len(s.Endpoints), 1)
|
||||
assert.Equal(c, s.Endpoints[0], "127.0.0.1:1234")
|
||||
assert.Equal(c, d.path, "path/"+defaultDiscoveryPath)
|
||||
|
||||
storeMock = &FakeStore{
|
||||
Endpoints: []string{"127.0.0.1:1234", "127.0.0.2:1234", "127.0.0.3:1234"},
|
||||
}
|
||||
d = &Discovery{backend: store.CONSUL}
|
||||
d.Initialize("127.0.0.1:1234,127.0.0.2:1234,127.0.0.3:1234/path", 0, 0, nil)
|
||||
d.store = storeMock
|
||||
|
||||
s = d.store.(*FakeStore)
|
||||
assert.Equal(c, len(s.Endpoints), 3)
|
||||
assert.Equal(c, s.Endpoints[0], "127.0.0.1:1234")
|
||||
assert.Equal(c, s.Endpoints[1], "127.0.0.2:1234")
|
||||
assert.Equal(c, s.Endpoints[2], "127.0.0.3:1234")
|
||||
|
||||
assert.Equal(c, d.path, "path/"+defaultDiscoveryPath)
|
||||
}
|
||||
|
||||
// Extremely limited mock store so we can test initialization
|
||||
type Mock struct {
|
||||
// Endpoints passed to InitializeMock
|
||||
Endpoints []string
|
||||
|
||||
// Options passed to InitializeMock
|
||||
Options *store.Config
|
||||
}
|
||||
|
||||
func NewMock(endpoints []string, options *store.Config) (store.Store, error) {
|
||||
s := &Mock{}
|
||||
s.Endpoints = endpoints
|
||||
s.Options = options
|
||||
return s, nil
|
||||
}
|
||||
func (s *Mock) Put(key string, value []byte, opts *store.WriteOptions) error {
|
||||
return errors.New("Put not supported")
|
||||
}
|
||||
func (s *Mock) Get(key string) (*store.KVPair, error) {
|
||||
return nil, errors.New("Get not supported")
|
||||
}
|
||||
func (s *Mock) Delete(key string) error {
|
||||
return errors.New("Delete not supported")
|
||||
}
|
||||
|
||||
// Exists mock
|
||||
func (s *Mock) Exists(key string) (bool, error) {
|
||||
return false, errors.New("Exists not supported")
|
||||
}
|
||||
|
||||
// Watch mock
|
||||
func (s *Mock) Watch(key string, stopCh <-chan struct{}) (<-chan *store.KVPair, error) {
|
||||
return nil, errors.New("Watch not supported")
|
||||
}
|
||||
|
||||
// WatchTree mock
|
||||
func (s *Mock) WatchTree(prefix string, stopCh <-chan struct{}) (<-chan []*store.KVPair, error) {
|
||||
return nil, errors.New("WatchTree not supported")
|
||||
}
|
||||
|
||||
// NewLock mock
|
||||
func (s *Mock) NewLock(key string, options *store.LockOptions) (store.Locker, error) {
|
||||
return nil, errors.New("NewLock not supported")
|
||||
}
|
||||
|
||||
// List mock
|
||||
func (s *Mock) List(prefix string) ([]*store.KVPair, error) {
|
||||
return nil, errors.New("List not supported")
|
||||
}
|
||||
|
||||
// DeleteTree mock
|
||||
func (s *Mock) DeleteTree(prefix string) error {
|
||||
return errors.New("DeleteTree not supported")
|
||||
}
|
||||
|
||||
// AtomicPut mock
|
||||
func (s *Mock) AtomicPut(key string, value []byte, previous *store.KVPair, opts *store.WriteOptions) (bool, *store.KVPair, error) {
|
||||
return false, nil, errors.New("AtomicPut not supported")
|
||||
}
|
||||
|
||||
// AtomicDelete mock
|
||||
func (s *Mock) AtomicDelete(key string, previous *store.KVPair) (bool, error) {
|
||||
return false, errors.New("AtomicDelete not supported")
|
||||
}
|
||||
|
||||
// Close mock
|
||||
func (s *Mock) Close() {
|
||||
}
|
||||
|
||||
func (ds *DiscoverySuite) TestInitializeWithCerts(c *testing.T) {
|
||||
cert := `-----BEGIN CERTIFICATE-----
|
||||
MIIDCDCCAfKgAwIBAgIICifG7YeiQOEwCwYJKoZIhvcNAQELMBIxEDAOBgNVBAMT
|
||||
B1Rlc3QgQ0EwHhcNMTUxMDAxMjMwMDAwWhcNMjAwOTI5MjMwMDAwWjASMRAwDgYD
|
||||
VQQDEwdUZXN0IENBMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA1wRC
|
||||
O+flnLTK5ImjTurNRHwSejuqGbc4CAvpB0hS+z0QlSs4+zE9h80aC4hz+6caRpds
|
||||
+J908Q+RvAittMHbpc7VjbZP72G6fiXk7yPPl6C10HhRSoSi3nY+B7F2E8cuz14q
|
||||
V2e+ejhWhSrBb/keyXpcyjoW1BOAAJ2TIclRRkICSCZrpXUyXxAvzXfpFXo1RhSb
|
||||
UywN11pfiCQzDUN7sPww9UzFHuAHZHoyfTr27XnJYVUerVYrCPq8vqfn//01qz55
|
||||
Xs0hvzGdlTFXhuabFtQnKFH5SNwo/fcznhB7rePOwHojxOpXTBepUCIJLbtNnWFT
|
||||
V44t9gh5IqIWtoBReQIDAQABo2YwZDAOBgNVHQ8BAf8EBAMCAAYwEgYDVR0TAQH/
|
||||
BAgwBgEB/wIBAjAdBgNVHQ4EFgQUZKUI8IIjIww7X/6hvwggQK4bD24wHwYDVR0j
|
||||
BBgwFoAUZKUI8IIjIww7X/6hvwggQK4bD24wCwYJKoZIhvcNAQELA4IBAQDES2cz
|
||||
7sCQfDCxCIWH7X8kpi/JWExzUyQEJ0rBzN1m3/x8ySRxtXyGekimBqQwQdFqlwMI
|
||||
xzAQKkh3ue8tNSzRbwqMSyH14N1KrSxYS9e9szJHfUasoTpQGPmDmGIoRJuq1h6M
|
||||
ej5x1SCJ7GWCR6xEXKUIE9OftXm9TdFzWa7Ja3OHz/mXteii8VXDuZ5ACq6EE5bY
|
||||
8sP4gcICfJ5fTrpTlk9FIqEWWQrCGa5wk95PGEj+GJpNogjXQ97wVoo/Y3p1brEn
|
||||
t5zjN9PAq4H1fuCMdNNA+p1DHNwd+ELTxcMAnb2ajwHvV6lKPXutrTFc4umJToBX
|
||||
FpTxDmJHEV4bzUzh
|
||||
-----END CERTIFICATE-----
|
||||
`
|
||||
key := `-----BEGIN RSA PRIVATE KEY-----
|
||||
MIIEpQIBAAKCAQEA1wRCO+flnLTK5ImjTurNRHwSejuqGbc4CAvpB0hS+z0QlSs4
|
||||
+zE9h80aC4hz+6caRpds+J908Q+RvAittMHbpc7VjbZP72G6fiXk7yPPl6C10HhR
|
||||
SoSi3nY+B7F2E8cuz14qV2e+ejhWhSrBb/keyXpcyjoW1BOAAJ2TIclRRkICSCZr
|
||||
pXUyXxAvzXfpFXo1RhSbUywN11pfiCQzDUN7sPww9UzFHuAHZHoyfTr27XnJYVUe
|
||||
rVYrCPq8vqfn//01qz55Xs0hvzGdlTFXhuabFtQnKFH5SNwo/fcznhB7rePOwHoj
|
||||
xOpXTBepUCIJLbtNnWFTV44t9gh5IqIWtoBReQIDAQABAoIBAHSWipORGp/uKFXj
|
||||
i/mut776x8ofsAxhnLBARQr93ID+i49W8H7EJGkOfaDjTICYC1dbpGrri61qk8sx
|
||||
qX7p3v/5NzKwOIfEpirgwVIqSNYe/ncbxnhxkx6tXtUtFKmEx40JskvSpSYAhmmO
|
||||
1XSx0E/PWaEN/nLgX/f1eWJIlxlQkk3QeqL+FGbCXI48DEtlJ9+MzMu4pAwZTpj5
|
||||
5qtXo5JJ0jRGfJVPAOznRsYqv864AhMdMIWguzk6EGnbaCWwPcfcn+h9a5LMdony
|
||||
MDHfBS7bb5tkF3+AfnVY3IBMVx7YlsD9eAyajlgiKu4zLbwTRHjXgShy+4Oussz0
|
||||
ugNGnkECgYEA/hi+McrZC8C4gg6XqK8+9joD8tnyDZDz88BQB7CZqABUSwvjDqlP
|
||||
L8hcwo/lzvjBNYGkqaFPUICGWKjeCtd8pPS2DCVXxDQX4aHF1vUur0uYNncJiV3N
|
||||
XQz4Iemsa6wnKf6M67b5vMXICw7dw0HZCdIHD1hnhdtDz0uVpeevLZ8CgYEA2KCT
|
||||
Y43lorjrbCgMqtlefkr3GJA9dey+hTzCiWEOOqn9RqGoEGUday0sKhiLofOgmN2B
|
||||
LEukpKIey8s+Q/cb6lReajDVPDsMweX8i7hz3Wa4Ugp4Xa5BpHqu8qIAE2JUZ7bU
|
||||
t88aQAYE58pUF+/Lq1QzAQdrjjzQBx6SrBxieecCgYEAvukoPZEC8mmiN1VvbTX+
|
||||
QFHmlZha3QaDxChB+QUe7bMRojEUL/fVnzkTOLuVFqSfxevaI/km9n0ac5KtAchV
|
||||
xjp2bTnBb5EUQFqjopYktWA+xO07JRJtMfSEmjZPbbay1kKC7rdTfBm961EIHaRj
|
||||
xZUf6M+rOE8964oGrdgdLlECgYEA046GQmx6fh7/82FtdZDRQp9tj3SWQUtSiQZc
|
||||
qhO59Lq8mjUXz+MgBuJXxkiwXRpzlbaFB0Bca1fUoYw8o915SrDYf/Zu2OKGQ/qa
|
||||
V81sgiVmDuEgycR7YOlbX6OsVUHrUlpwhY3hgfMe6UtkMvhBvHF/WhroBEIJm1pV
|
||||
PXZ/CbMCgYEApNWVktFBjOaYfY6SNn4iSts1jgsQbbpglg3kT7PLKjCAhI6lNsbk
|
||||
dyT7ut01PL6RaW4SeQWtrJIVQaM6vF3pprMKqlc5XihOGAmVqH7rQx9rtQB5TicL
|
||||
BFrwkQE4HQtQBV60hYQUzzlSk44VFDz+jxIEtacRHaomDRh2FtOTz+I=
|
||||
-----END RSA PRIVATE KEY-----
|
||||
`
|
||||
certFile, err := os.CreateTemp("", "cert")
|
||||
assert.Assert(c, err == nil)
|
||||
defer os.Remove(certFile.Name())
|
||||
certFile.Write([]byte(cert))
|
||||
certFile.Close()
|
||||
keyFile, err := os.CreateTemp("", "key")
|
||||
assert.Assert(c, err == nil)
|
||||
defer os.Remove(keyFile.Name())
|
||||
keyFile.Write([]byte(key))
|
||||
keyFile.Close()
|
||||
|
||||
libkv.AddStore("mock", NewMock)
|
||||
d := &Discovery{backend: "mock"}
|
||||
err = d.Initialize("127.0.0.3:1234", 0, 0, map[string]string{
|
||||
"kv.cacertfile": certFile.Name(),
|
||||
"kv.certfile": certFile.Name(),
|
||||
"kv.keyfile": keyFile.Name(),
|
||||
})
|
||||
assert.Assert(c, err == nil)
|
||||
s := d.store.(*Mock)
|
||||
assert.Assert(c, s.Options.TLS != nil)
|
||||
assert.Assert(c, s.Options.TLS.RootCAs != nil)
|
||||
assert.Equal(c, len(s.Options.TLS.Certificates), 1)
|
||||
}
|
||||
|
||||
func (ds *DiscoverySuite) TestWatch(c *testing.T) {
|
||||
mockCh := make(chan []*store.KVPair)
|
||||
|
||||
storeMock := &FakeStore{
|
||||
Endpoints: []string{"127.0.0.1:1234"},
|
||||
mockKVChan: mockCh,
|
||||
}
|
||||
|
||||
d := &Discovery{backend: store.CONSUL}
|
||||
d.Initialize("127.0.0.1:1234/path", 0, 0, nil)
|
||||
d.store = storeMock
|
||||
|
||||
expected := discovery.Entries{
|
||||
&discovery.Entry{Host: "1.1.1.1", Port: "1111"},
|
||||
&discovery.Entry{Host: "2.2.2.2", Port: "2222"},
|
||||
}
|
||||
kvs := []*store.KVPair{
|
||||
{Key: path.Join("path", defaultDiscoveryPath, "1.1.1.1"), Value: []byte("1.1.1.1:1111")},
|
||||
{Key: path.Join("path", defaultDiscoveryPath, "2.2.2.2"), Value: []byte("2.2.2.2:2222")},
|
||||
}
|
||||
|
||||
stopCh := make(chan struct{})
|
||||
ch, errCh := d.Watch(stopCh)
|
||||
|
||||
// It should fire an error since the first WatchTree call failed.
|
||||
assert.ErrorContains(c, <-errCh, "test error")
|
||||
// We have to drain the error channel otherwise Watch will get stuck.
|
||||
go func() {
|
||||
for range errCh {
|
||||
}
|
||||
}()
|
||||
|
||||
// Push the entries into the store channel and make sure discovery emits.
|
||||
mockCh <- kvs
|
||||
assert.DeepEqual(c, <-ch, expected)
|
||||
|
||||
// Add a new entry.
|
||||
expected = append(expected, &discovery.Entry{Host: "3.3.3.3", Port: "3333"})
|
||||
kvs = append(kvs, &store.KVPair{Key: path.Join("path", defaultDiscoveryPath, "3.3.3.3"), Value: []byte("3.3.3.3:3333")})
|
||||
mockCh <- kvs
|
||||
assert.DeepEqual(c, <-ch, expected)
|
||||
|
||||
close(mockCh)
|
||||
// Give it enough time to call WatchTree.
|
||||
time.Sleep(3 * time.Second)
|
||||
|
||||
// Stop and make sure it closes all channels.
|
||||
close(stopCh)
|
||||
assert.Assert(c, <-ch == nil)
|
||||
assert.Assert(c, <-errCh == nil)
|
||||
}
|
||||
|
||||
// FakeStore implements store.Store methods. It mocks all store
|
||||
// function in a simple, naive way.
|
||||
type FakeStore struct {
|
||||
Endpoints []string
|
||||
Options *store.Config
|
||||
mockKVChan <-chan []*store.KVPair
|
||||
|
||||
watchTreeCallCount int
|
||||
}
|
||||
|
||||
func (s *FakeStore) Put(key string, value []byte, options *store.WriteOptions) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *FakeStore) Get(key string) (*store.KVPair, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (s *FakeStore) Delete(key string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *FakeStore) Exists(key string) (bool, error) {
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (s *FakeStore) Watch(key string, stopCh <-chan struct{}) (<-chan *store.KVPair, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// WatchTree will fail the first time, and return the mockKVchan afterwards.
|
||||
// This is the behavior we need for testing.. If we need 'moar', should update this.
|
||||
func (s *FakeStore) WatchTree(directory string, stopCh <-chan struct{}) (<-chan []*store.KVPair, error) {
|
||||
if s.watchTreeCallCount == 0 {
|
||||
s.watchTreeCallCount = 1
|
||||
return nil, errors.New("test error")
|
||||
}
|
||||
// First calls error
|
||||
return s.mockKVChan, nil
|
||||
}
|
||||
|
||||
func (s *FakeStore) NewLock(key string, options *store.LockOptions) (store.Locker, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (s *FakeStore) List(directory string) ([]*store.KVPair, error) {
|
||||
return []*store.KVPair{}, nil
|
||||
}
|
||||
|
||||
func (s *FakeStore) DeleteTree(directory string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *FakeStore) AtomicPut(key string, value []byte, previous *store.KVPair, options *store.WriteOptions) (bool, *store.KVPair, error) {
|
||||
return true, nil, nil
|
||||
}
|
||||
|
||||
func (s *FakeStore) AtomicDelete(key string, previous *store.KVPair) (bool, error) {
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (s *FakeStore) Close() {
|
||||
}
|
||||
|
|
@ -1,93 +0,0 @@
|
|||
package memory // import "github.com/docker/docker/pkg/discovery/memory"
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/docker/docker/pkg/discovery"
|
||||
)
|
||||
|
||||
// Discovery implements a discovery backend that keeps
|
||||
// data in memory.
|
||||
type Discovery struct {
|
||||
heartbeat time.Duration
|
||||
values []string
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
func init() {
|
||||
Init()
|
||||
}
|
||||
|
||||
// Init registers the memory backend on demand.
|
||||
func Init() {
|
||||
discovery.Register("memory", &Discovery{})
|
||||
}
|
||||
|
||||
// Initialize sets the heartbeat for the memory backend.
|
||||
func (s *Discovery) Initialize(_ string, heartbeat time.Duration, _ time.Duration, _ map[string]string) error {
|
||||
s.heartbeat = heartbeat
|
||||
s.values = make([]string, 0)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Watch sends periodic discovery updates to a channel.
|
||||
func (s *Discovery) Watch(stopCh <-chan struct{}) (<-chan discovery.Entries, <-chan error) {
|
||||
ch := make(chan discovery.Entries, 1)
|
||||
errCh := make(chan error, 1)
|
||||
ticker := time.NewTicker(s.heartbeat)
|
||||
|
||||
go func() {
|
||||
defer close(errCh)
|
||||
defer close(ch)
|
||||
|
||||
// Send the initial entries if available.
|
||||
var currentEntries discovery.Entries
|
||||
var err error
|
||||
|
||||
s.mu.Lock()
|
||||
if len(s.values) > 0 {
|
||||
currentEntries, err = discovery.CreateEntries(s.values)
|
||||
}
|
||||
s.mu.Unlock()
|
||||
|
||||
if err != nil {
|
||||
errCh <- err
|
||||
} else if currentEntries != nil {
|
||||
ch <- currentEntries
|
||||
}
|
||||
|
||||
// Periodically send updates.
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
s.mu.Lock()
|
||||
newEntries, err := discovery.CreateEntries(s.values)
|
||||
s.mu.Unlock()
|
||||
if err != nil {
|
||||
errCh <- err
|
||||
continue
|
||||
}
|
||||
|
||||
// Check if the file has really changed.
|
||||
if !newEntries.Equals(currentEntries) {
|
||||
ch <- newEntries
|
||||
}
|
||||
currentEntries = newEntries
|
||||
case <-stopCh:
|
||||
ticker.Stop()
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
return ch, errCh
|
||||
}
|
||||
|
||||
// Register adds a new address to the discovery.
|
||||
func (s *Discovery) Register(addr string) error {
|
||||
s.mu.Lock()
|
||||
s.values = append(s.values, addr)
|
||||
s.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
|
@ -1,49 +0,0 @@
|
|||
package memory // import "github.com/docker/docker/pkg/discovery/memory"
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/docker/docker/internal/test/suite"
|
||||
"github.com/docker/docker/pkg/discovery"
|
||||
"gotest.tools/v3/assert"
|
||||
)
|
||||
|
||||
// Hook up gocheck into the "go test" runner.
|
||||
func Test(t *testing.T) {
|
||||
suite.Run(t, &discoverySuite{})
|
||||
}
|
||||
|
||||
type discoverySuite struct{}
|
||||
|
||||
func (s *discoverySuite) TestWatch(c *testing.T) {
|
||||
d := &Discovery{}
|
||||
d.Initialize("foo", 1000, 0, nil)
|
||||
stopCh := make(chan struct{})
|
||||
ch, errCh := d.Watch(stopCh)
|
||||
|
||||
// We have to drain the error channel otherwise Watch will get stuck.
|
||||
go func() {
|
||||
for range errCh {
|
||||
}
|
||||
}()
|
||||
|
||||
expected := discovery.Entries{
|
||||
&discovery.Entry{Host: "1.1.1.1", Port: "1111"},
|
||||
}
|
||||
|
||||
assert.Assert(c, d.Register("1.1.1.1:1111") == nil)
|
||||
assert.DeepEqual(c, <-ch, expected)
|
||||
|
||||
expected = discovery.Entries{
|
||||
&discovery.Entry{Host: "1.1.1.1", Port: "1111"},
|
||||
&discovery.Entry{Host: "2.2.2.2", Port: "2222"},
|
||||
}
|
||||
|
||||
assert.Assert(c, d.Register("2.2.2.2:2222") == nil)
|
||||
assert.DeepEqual(c, <-ch, expected)
|
||||
|
||||
// Stop and make sure it closes all channels.
|
||||
close(stopCh)
|
||||
assert.Assert(c, <-ch == nil)
|
||||
assert.Assert(c, <-errCh == nil)
|
||||
}
|
||||
|
|
@ -1,54 +0,0 @@
|
|||
package nodes // import "github.com/docker/docker/pkg/discovery/nodes"
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/docker/docker/pkg/discovery"
|
||||
)
|
||||
|
||||
// Discovery is exported
|
||||
type Discovery struct {
|
||||
entries discovery.Entries
|
||||
}
|
||||
|
||||
func init() {
|
||||
Init()
|
||||
}
|
||||
|
||||
// Init is exported
|
||||
func Init() {
|
||||
discovery.Register("nodes", &Discovery{})
|
||||
}
|
||||
|
||||
// Initialize is exported
|
||||
func (s *Discovery) Initialize(uris string, _ time.Duration, _ time.Duration, _ map[string]string) error {
|
||||
for _, input := range strings.Split(uris, ",") {
|
||||
for _, ip := range discovery.Generate(input) {
|
||||
entry, err := discovery.NewEntry(ip)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%s, please check you are using the correct discovery (missing token:// ?)", err.Error())
|
||||
}
|
||||
s.entries = append(s.entries, entry)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Watch is exported
|
||||
func (s *Discovery) Watch(stopCh <-chan struct{}) (<-chan discovery.Entries, <-chan error) {
|
||||
ch := make(chan discovery.Entries, 1)
|
||||
go func() {
|
||||
defer close(ch)
|
||||
ch <- s.entries
|
||||
<-stopCh
|
||||
}()
|
||||
return ch, nil
|
||||
}
|
||||
|
||||
// Register is exported
|
||||
func (s *Discovery) Register(addr string) error {
|
||||
return discovery.ErrNotImplemented
|
||||
}
|
||||
|
|
@ -1,51 +0,0 @@
|
|||
package nodes // import "github.com/docker/docker/pkg/discovery/nodes"
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/docker/docker/internal/test/suite"
|
||||
"github.com/docker/docker/pkg/discovery"
|
||||
"gotest.tools/v3/assert"
|
||||
)
|
||||
|
||||
// Hook up gocheck into the "go test" runner.
|
||||
func Test(t *testing.T) {
|
||||
suite.Run(t, &DiscoverySuite{})
|
||||
}
|
||||
|
||||
type DiscoverySuite struct{}
|
||||
|
||||
func (s *DiscoverySuite) TestInitialize(c *testing.T) {
|
||||
d := &Discovery{}
|
||||
d.Initialize("1.1.1.1:1111,2.2.2.2:2222", 0, 0, nil)
|
||||
assert.Equal(c, len(d.entries), 2)
|
||||
assert.Equal(c, d.entries[0].String(), "1.1.1.1:1111")
|
||||
assert.Equal(c, d.entries[1].String(), "2.2.2.2:2222")
|
||||
}
|
||||
|
||||
func (s *DiscoverySuite) TestInitializeWithPattern(c *testing.T) {
|
||||
d := &Discovery{}
|
||||
d.Initialize("1.1.1.[1:2]:1111,2.2.2.[2:4]:2222", 0, 0, nil)
|
||||
assert.Equal(c, len(d.entries), 5)
|
||||
assert.Equal(c, d.entries[0].String(), "1.1.1.1:1111")
|
||||
assert.Equal(c, d.entries[1].String(), "1.1.1.2:1111")
|
||||
assert.Equal(c, d.entries[2].String(), "2.2.2.2:2222")
|
||||
assert.Equal(c, d.entries[3].String(), "2.2.2.3:2222")
|
||||
assert.Equal(c, d.entries[4].String(), "2.2.2.4:2222")
|
||||
}
|
||||
|
||||
func (s *DiscoverySuite) TestWatch(c *testing.T) {
|
||||
d := &Discovery{}
|
||||
d.Initialize("1.1.1.1:1111,2.2.2.2:2222", 0, 0, nil)
|
||||
expected := discovery.Entries{
|
||||
&discovery.Entry{Host: "1.1.1.1", Port: "1111"},
|
||||
&discovery.Entry{Host: "2.2.2.2", Port: "2222"},
|
||||
}
|
||||
ch, _ := d.Watch(nil)
|
||||
assert.Equal(c, expected.Equals(<-ch), true)
|
||||
}
|
||||
|
||||
func (s *DiscoverySuite) TestRegister(c *testing.T) {
|
||||
d := &Discovery{}
|
||||
assert.Assert(c, d.Register("0.0.0.0") != nil)
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue