open-consul/command/agent/scada.go

194 lines
4.3 KiB
Go
Raw Normal View History

2015-02-05 02:06:36 +00:00
package agent
import (
"crypto/tls"
2015-02-05 02:36:35 +00:00
"errors"
2015-02-05 02:06:36 +00:00
"fmt"
"io"
2015-02-05 02:17:45 +00:00
"net"
2015-02-06 22:10:01 +00:00
"os"
2015-02-05 02:17:45 +00:00
"strconv"
2015-02-05 02:36:35 +00:00
"sync"
"time"
2015-02-05 02:06:36 +00:00
"github.com/hashicorp/scada-client"
)
const (
// providerService is the service name we use
providerService = "consul"
// resourceType is the type of resource we represent
// when connecting to SCADA
resourceType = "infrastructures"
)
// 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.AtlasJoin),
2015-02-05 02:17:45 +00:00
"datacenter": c.Datacenter,
"server": strconv.FormatBool(c.Server),
2015-02-05 02:06:36 +00:00
},
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,
},
2015-08-27 18:08:01 +00:00
Endpoint: c.AtlasEndpoint,
ResourceGroup: c.AtlasInfrastructure,
2015-02-05 02:06:36 +00:00
Token: c.AtlasToken,
}
}
// NewProvider creates a new SCADA provider using the
2015-02-19 00:54:44 +00:00
// given configuration. Requests for the HTTP capability
// are passed off to the listener that is returned.
2015-02-05 02:17:45 +00:00
func NewProvider(c *Config, logOutput io.Writer) (*client.Provider, net.Listener, error) {
2015-02-05 02:06:36 +00:00
// Get the configuration of the provider
config := ProviderConfig(c)
2015-02-07 02:53:02 +00:00
config.LogOutput = logOutput
2015-02-05 02:06:36 +00:00
2015-02-06 22:10:01 +00:00
// SCADA_INSECURE env variable is used for testing to disable
// TLS certificate verification.
if os.Getenv("SCADA_INSECURE") != "" {
config.TLSConfig = &tls.Config{
InsecureSkipVerify: true,
}
2015-02-05 02:06:36 +00:00
}
2015-02-05 02:36:35 +00:00
// Create an HTTP listener and handler
list := newScadaListener(c.AtlasInfrastructure)
2015-02-05 02:36:35 +00:00
config.Handlers["http"] = func(capability string, meta map[string]string,
conn io.ReadWriteCloser) error {
return list.PushRWC(conn)
}
2015-02-05 02:06:36 +00:00
// Create the provider
2015-02-05 02:17:45 +00:00
provider, err := client.NewProvider(config)
if err != nil {
2015-02-05 02:36:35 +00:00
list.Close()
2015-02-05 02:17:45 +00:00
return nil, nil, err
}
2015-02-05 02:36:35 +00:00
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
2015-02-06 22:27:11 +00:00
func newScadaListener(infra string) *scadaListener {
2015-02-05 02:36:35 +00:00
l := &scadaListener{
2015-02-06 22:27:11 +00:00
addr: &scadaAddr{infra},
2015-02-05 02:36:35 +00:00
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)
}
2015-09-15 12:22:08 +00:00
// Push is used to add a connection to the queue
2015-02-05 02:36:35 +00:00
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")
2015-02-05 02:36:35 +00:00
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 {
2015-02-06 22:27:11 +00:00
infra string
2015-02-05 02:36:35 +00:00
}
func (s *scadaAddr) Network() string {
return "SCADA"
}
func (s *scadaAddr) String() string {
2015-02-06 22:27:11 +00:00
return fmt.Sprintf("SCADA::Atlas::%s", s.infra)
2015-02-05 02:36:35 +00:00
}
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")
2015-02-05 02:06:36 +00:00
}