mirror of
https://github.com/kemko/nomad.git
synced 2026-01-08 19:35:41 +03:00
Merge pull request #1227 from hashicorp/upstream-scada-provider
Use upstream high-level SCADA provider
This commit is contained in:
Alex Dadgar
@@ -21,7 +21,7 @@ import (
|
||||
"github.com/hashicorp/logutils"
|
||||
"github.com/hashicorp/nomad/helper/flag-slice"
|
||||
"github.com/hashicorp/nomad/helper/gated-writer"
|
||||
scada "github.com/hashicorp/scada-client"
|
||||
"github.com/hashicorp/scada-client/scada"
|
||||
"github.com/mitchellh/cli"
|
||||
)
|
||||
|
||||
@@ -622,7 +622,26 @@ func (c *Command) setupSCADA(config *Config) error {
|
||||
|
||||
// Create the new provider and listener
|
||||
c.Ui.Output("Connecting to Atlas: " + config.Atlas.Infrastructure)
|
||||
provider, list, err := NewProvider(config, c.logOutput)
|
||||
|
||||
scadaConfig := &scada.Config{
|
||||
Service: "nomad",
|
||||
Version: fmt.Sprintf("%s%s", config.Version, config.VersionPrerelease),
|
||||
ResourceType: "nomad-cluster",
|
||||
Meta: map[string]string{
|
||||
"auto-join": strconv.FormatBool(config.Atlas.Join),
|
||||
"region": config.Region,
|
||||
"datacenter": config.Datacenter,
|
||||
"client": strconv.FormatBool(config.Client != nil && config.Client.Enabled),
|
||||
"server": strconv.FormatBool(config.Server != nil && config.Server.Enabled),
|
||||
},
|
||||
Atlas: scada.AtlasConfig{
|
||||
Endpoint: config.Atlas.Endpoint,
|
||||
Infrastructure: config.Atlas.Infrastructure,
|
||||
Token: config.Atlas.Token,
|
||||
},
|
||||
}
|
||||
|
||||
provider, list, err := scada.NewHTTPProvider(scadaConfig, c.logOutput)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -1,195 +0,0 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"os"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/hashicorp/scada-client"
|
||||
)
|
||||
|
||||
const (
|
||||
// providerService is the service name we use
|
||||
providerService = "nomad"
|
||||
|
||||
// resourceType is the type of resource we represent
|
||||
// when connecting to SCADA
|
||||
resourceType = "nomad-cluster"
|
||||
)
|
||||
|
||||
// ProviderService returns the service information for the provider
|
||||
func ProviderService(c *Config) *client.ProviderService {
|
||||
return &client.ProviderService{
|
||||
Service: providerService,
|
||||
ServiceVersion: fmt.Sprintf("%s%s", c.Version, c.VersionPrerelease),
|
||||
Capabilities: map[string]int{
|
||||
"http": 1,
|
||||
},
|
||||
Meta: map[string]string{
|
||||
"auto-join": strconv.FormatBool(c.Atlas.Join),
|
||||
"region": c.Region,
|
||||
"datacenter": c.Datacenter,
|
||||
"client": strconv.FormatBool(c.Client != nil && c.Client.Enabled),
|
||||
"server": strconv.FormatBool(c.Server != nil && c.Server.Enabled),
|
||||
},
|
||||
ResourceType: resourceType,
|
||||
}
|
||||
}
|
||||
|
||||
// ProviderConfig returns the configuration for the SCADA provider
|
||||
func ProviderConfig(c *Config) *client.ProviderConfig {
|
||||
return &client.ProviderConfig{
|
||||
Service: ProviderService(c),
|
||||
Handlers: map[string]client.CapabilityProvider{
|
||||
"http": nil,
|
||||
},
|
||||
Endpoint: c.Atlas.Endpoint,
|
||||
ResourceGroup: c.Atlas.Infrastructure,
|
||||
Token: c.Atlas.Token,
|
||||
}
|
||||
}
|
||||
|
||||
// NewProvider creates a new SCADA provider using the
|
||||
// given configuration. Requests for the HTTP capability
|
||||
// are passed off to the listener that is returned.
|
||||
func NewProvider(c *Config, logOutput io.Writer) (*client.Provider, net.Listener, error) {
|
||||
// Get the configuration of the provider
|
||||
config := ProviderConfig(c)
|
||||
config.LogOutput = logOutput
|
||||
|
||||
// SCADA_INSECURE env variable is used for testing to disable
|
||||
// TLS certificate verification.
|
||||
if os.Getenv("SCADA_INSECURE") != "" {
|
||||
config.TLSConfig = &tls.Config{
|
||||
InsecureSkipVerify: true,
|
||||
}
|
||||
}
|
||||
|
||||
// Create an HTTP listener and handler
|
||||
list := newScadaListener(c.Atlas.Infrastructure)
|
||||
config.Handlers["http"] = func(capability string, meta map[string]string,
|
||||
conn io.ReadWriteCloser) error {
|
||||
return list.PushRWC(conn)
|
||||
}
|
||||
|
||||
// Create the provider
|
||||
provider, err := client.NewProvider(config)
|
||||
if err != nil {
|
||||
list.Close()
|
||||
return nil, nil, err
|
||||
}
|
||||
return provider, list, nil
|
||||
}
|
||||
|
||||
// scadaListener is used to return a net.Listener for
|
||||
// incoming SCADA connections
|
||||
type scadaListener struct {
|
||||
addr *scadaAddr
|
||||
pending chan net.Conn
|
||||
|
||||
closed bool
|
||||
closedCh chan struct{}
|
||||
l sync.Mutex
|
||||
}
|
||||
|
||||
// newScadaListener returns a new listener
|
||||
func newScadaListener(infra string) *scadaListener {
|
||||
l := &scadaListener{
|
||||
addr: &scadaAddr{infra},
|
||||
pending: make(chan net.Conn),
|
||||
closedCh: make(chan struct{}),
|
||||
}
|
||||
return l
|
||||
}
|
||||
|
||||
// PushRWC is used to push a io.ReadWriteCloser as a net.Conn
|
||||
func (s *scadaListener) PushRWC(conn io.ReadWriteCloser) error {
|
||||
// Check if this already implements net.Conn
|
||||
if nc, ok := conn.(net.Conn); ok {
|
||||
return s.Push(nc)
|
||||
}
|
||||
|
||||
// Wrap to implement the interface
|
||||
wrapped := &scadaRWC{conn, s.addr}
|
||||
return s.Push(wrapped)
|
||||
}
|
||||
|
||||
// Push is used to add a connection to the queu
|
||||
func (s *scadaListener) Push(conn net.Conn) error {
|
||||
select {
|
||||
case s.pending <- conn:
|
||||
return nil
|
||||
case <-time.After(time.Second):
|
||||
return fmt.Errorf("accept timed out")
|
||||
case <-s.closedCh:
|
||||
return fmt.Errorf("scada listener closed")
|
||||
}
|
||||
}
|
||||
|
||||
func (s *scadaListener) Accept() (net.Conn, error) {
|
||||
select {
|
||||
case conn := <-s.pending:
|
||||
return conn, nil
|
||||
case <-s.closedCh:
|
||||
return nil, fmt.Errorf("scada listener closed")
|
||||
}
|
||||
}
|
||||
|
||||
func (s *scadaListener) Close() error {
|
||||
s.l.Lock()
|
||||
defer s.l.Unlock()
|
||||
if s.closed {
|
||||
return nil
|
||||
}
|
||||
s.closed = true
|
||||
close(s.closedCh)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *scadaListener) Addr() net.Addr {
|
||||
return s.addr
|
||||
}
|
||||
|
||||
// scadaAddr is used to return a net.Addr for SCADA
|
||||
type scadaAddr struct {
|
||||
infra string
|
||||
}
|
||||
|
||||
func (s *scadaAddr) Network() string {
|
||||
return "SCADA"
|
||||
}
|
||||
|
||||
func (s *scadaAddr) String() string {
|
||||
return fmt.Sprintf("SCADA::Atlas::%s", s.infra)
|
||||
}
|
||||
|
||||
type scadaRWC struct {
|
||||
io.ReadWriteCloser
|
||||
addr *scadaAddr
|
||||
}
|
||||
|
||||
func (s *scadaRWC) LocalAddr() net.Addr {
|
||||
return s.addr
|
||||
}
|
||||
|
||||
func (s *scadaRWC) RemoteAddr() net.Addr {
|
||||
return s.addr
|
||||
}
|
||||
|
||||
func (s *scadaRWC) SetDeadline(t time.Time) error {
|
||||
return errors.New("SCADA.Conn does not support deadlines")
|
||||
}
|
||||
|
||||
func (s *scadaRWC) SetReadDeadline(t time.Time) error {
|
||||
return errors.New("SCADA.Conn does not support deadlines")
|
||||
}
|
||||
|
||||
func (s *scadaRWC) SetWriteDeadline(t time.Time) error {
|
||||
return errors.New("SCADA.Conn does not support deadlines")
|
||||
}
|
||||
@@ -1,112 +0,0 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"net"
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
"github.com/hashicorp/scada-client"
|
||||
)
|
||||
|
||||
func TestProviderService(t *testing.T) {
|
||||
conf := DefaultConfig()
|
||||
conf.Version = "0.5.0"
|
||||
conf.VersionPrerelease = "rc1"
|
||||
conf.Atlas = &AtlasConfig{}
|
||||
conf.Atlas.Join = true
|
||||
conf.Server.Enabled = true
|
||||
ps := ProviderService(conf)
|
||||
|
||||
expect := &client.ProviderService{
|
||||
Service: "nomad",
|
||||
ServiceVersion: "0.5.0rc1",
|
||||
Capabilities: map[string]int{
|
||||
"http": 1,
|
||||
},
|
||||
Meta: map[string]string{
|
||||
"auto-join": "true",
|
||||
"region": "global",
|
||||
"datacenter": "dc1",
|
||||
"client": "false",
|
||||
"server": "true",
|
||||
},
|
||||
ResourceType: "nomad-cluster",
|
||||
}
|
||||
|
||||
if !reflect.DeepEqual(ps, expect) {
|
||||
t.Fatalf("bad: %v", ps)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProviderConfig(t *testing.T) {
|
||||
conf := DefaultConfig()
|
||||
conf.Version = "0.5.0"
|
||||
conf.VersionPrerelease = "rc1"
|
||||
conf.Atlas = &AtlasConfig{}
|
||||
conf.Atlas.Join = true
|
||||
conf.Atlas.Infrastructure = "armon/test"
|
||||
conf.Atlas.Token = "foobarbaz"
|
||||
conf.Atlas.Endpoint = "foo.bar:1111"
|
||||
conf.Server.Enabled = true
|
||||
pc := ProviderConfig(conf)
|
||||
|
||||
expect := &client.ProviderConfig{
|
||||
Service: &client.ProviderService{
|
||||
Service: "nomad",
|
||||
ServiceVersion: "0.5.0rc1",
|
||||
Capabilities: map[string]int{
|
||||
"http": 1,
|
||||
},
|
||||
Meta: map[string]string{
|
||||
"auto-join": "true",
|
||||
"region": "global",
|
||||
"datacenter": "dc1",
|
||||
"client": "false",
|
||||
"server": "true",
|
||||
},
|
||||
ResourceType: "nomad-cluster",
|
||||
},
|
||||
Handlers: map[string]client.CapabilityProvider{
|
||||
"http": nil,
|
||||
},
|
||||
Endpoint: "foo.bar:1111",
|
||||
ResourceGroup: "armon/test",
|
||||
Token: "foobarbaz",
|
||||
}
|
||||
|
||||
if !reflect.DeepEqual(pc, expect) {
|
||||
t.Fatalf("bad: %v", pc)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSCADAListener(t *testing.T) {
|
||||
list := newScadaListener("armon/test")
|
||||
defer list.Close()
|
||||
|
||||
var raw interface{} = list
|
||||
_, ok := raw.(net.Listener)
|
||||
if !ok {
|
||||
t.Fatalf("bad")
|
||||
}
|
||||
|
||||
a, b := net.Pipe()
|
||||
defer a.Close()
|
||||
defer b.Close()
|
||||
|
||||
go list.Push(a)
|
||||
out, err := list.Accept()
|
||||
if err != nil {
|
||||
t.Fatalf("err: %v", err)
|
||||
}
|
||||
if out != a {
|
||||
t.Fatalf("bad")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSCADAAddr(t *testing.T) {
|
||||
var addr interface{} = &scadaAddr{"armon/test"}
|
||||
_, ok := addr.(net.Addr)
|
||||
if !ok {
|
||||
t.Fatalf("bad")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user