Merge pull request #4687 from hashicorp/connect-multidc-config

Connect multi-dc config

agent/agent.go

@@ -874,6 +874,7 @@ func (a *Agent) consulConfig() (*consul.Config, error) {
 	// todo(fs): these are now always set in the runtime config so we can simplify this
 	// todo(fs): or is there a reason to keep it like that?
 	base.Datacenter = a.config.Datacenter
+	base.PrimaryDatacenter = a.config.PrimaryDatacenter
 	base.DataDir = a.config.DataDir
 	base.NodeName = a.config.NodeName
 
@@ -1047,6 +1048,7 @@ func (a *Agent) consulConfig() (*consul.Config, error) {
 	// Copy the Connect CA bootstrap config
 	if a.config.ConnectEnabled {
 		base.ConnectEnabled = true
+		base.ConnectReplicationToken = a.config.ConnectReplicationToken
 
 		// Allow config to specify cluster_id provided it's a valid UUID. This is
 		// meant only for tests where a deterministic ID makes fixtures much simpler

agent/agent_endpoint.go

@@ -1152,6 +1152,9 @@ func (s *HTTPServer) AgentToken(resp http.ResponseWriter, req *http.Request) (in
 	case "acl_replication_token":
 		s.agent.tokens.UpdateACLReplicationToken(args.Token)
 
+	case "connect_replication_token":
+		s.agent.tokens.UpdateConnectReplicationToken(args.Token)
+
 	default:
 		resp.WriteHeader(http.StatusNotFound)
 		fmt.Fprintf(resp, "Token %q is unknown", target)

agent/config/builder.go

@@ -558,6 +558,15 @@ func (b *Builder) Build() (rt RuntimeConfig, err error) {
 		})
 	}
 
+	primaryDatacenter := strings.ToLower(b.stringVal(c.PrimaryDatacenter))
+	if c.ACLDatacenter != nil {
+		b.warn("The 'acl_datacenter' field is deprecated. Use the 'primary_datacenter' field instead.")
+
+		if primaryDatacenter == "" {
+			primaryDatacenter = strings.ToLower(b.stringVal(c.ACLDatacenter))
+		}
+	}
+
 	proxyDefaultExecMode := b.stringVal(c.Connect.ProxyDefaults.ExecMode)
 	proxyDefaultDaemonCommand := c.Connect.ProxyDefaults.DaemonCommand
 	proxyDefaultScriptCommand := c.Connect.ProxyDefaults.ScriptCommand
@@ -737,6 +746,7 @@ func (b *Builder) Build() (rt RuntimeConfig, err error) {
 		NodeName:                                b.nodeName(c.NodeName),
 		NonVotingServer:                         b.boolVal(c.NonVotingServer),
 		PidFile:                                 b.stringVal(c.PidFile),
+		PrimaryDatacenter:                       primaryDatacenter,
 		RPCAdvertiseAddr:                        rpcAdvertiseAddr,
 		RPCBindAddr:                             rpcBindAddr,
 		RPCHoldTimeout:                          b.durationVal("performance.rpc_hold_timeout", c.Performance.RPCHoldTimeout),

agent/config/config.go

@@ -212,6 +212,7 @@ type Config struct {
 	Performance                      Performance              `json:"performance,omitempty" hcl:"performance" mapstructure:"performance"`
 	PidFile                          *string                  `json:"pid_file,omitempty" hcl:"pid_file" mapstructure:"pid_file"`
 	Ports                            Ports                    `json:"ports,omitempty" hcl:"ports" mapstructure:"ports"`
+	PrimaryDatacenter                *string                  `json:"primary_datacenter,omitempty" hcl:"primary_datacenter" mapstructure:"primary_datacenter"`
 	RPCProtocol                      *int                     `json:"protocol,omitempty" hcl:"protocol" mapstructure:"protocol"`
 	RaftProtocol                     *int                     `json:"raft_protocol,omitempty" hcl:"raft_protocol" mapstructure:"raft_protocol"`
 	RaftSnapshotThreshold            *int                     `json:"raft_snapshot_threshold,omitempty" hcl:"raft_snapshot_threshold" mapstructure:"raft_snapshot_threshold"`
@@ -484,11 +485,12 @@ type Upstream struct {
 type Connect struct {
 	// Enabled opts the agent into connect. It should be set on all clients and
 	// servers in a cluster for correct connect operation.
-	Enabled       *bool                  `json:"enabled,omitempty" hcl:"enabled" mapstructure:"enabled"`
-	Proxy         ConnectProxy           `json:"proxy,omitempty" hcl:"proxy" mapstructure:"proxy"`
-	ProxyDefaults ConnectProxyDefaults   `json:"proxy_defaults,omitempty" hcl:"proxy_defaults" mapstructure:"proxy_defaults"`
-	CAProvider    *string                `json:"ca_provider,omitempty" hcl:"ca_provider" mapstructure:"ca_provider"`
-	CAConfig      map[string]interface{} `json:"ca_config,omitempty" hcl:"ca_config" mapstructure:"ca_config"`
+	Enabled          *bool                  `json:"enabled,omitempty" hcl:"enabled" mapstructure:"enabled"`
+	Proxy            ConnectProxy           `json:"proxy,omitempty" hcl:"proxy" mapstructure:"proxy"`
+	ProxyDefaults    ConnectProxyDefaults   `json:"proxy_defaults,omitempty" hcl:"proxy_defaults" mapstructure:"proxy_defaults"`
+	CAProvider       *string                `json:"ca_provider,omitempty" hcl:"ca_provider" mapstructure:"ca_provider"`
+	CAConfig         map[string]interface{} `json:"ca_config,omitempty" hcl:"ca_config" mapstructure:"ca_config"`
+	ReplicationToken *string                `json:"replication_token,omitempty" hcl:"replication_token" mapstructure:"replication_token"`
 }
 
 // ConnectProxy is the agent-global connect proxy configuration.

agent/config/runtime.go

@@ -503,6 +503,9 @@ type RuntimeConfig struct {
 	// ConnectCAConfig is the config to use for the CA provider.
 	ConnectCAConfig map[string]interface{}
 
+	// ConnectReplicationToken is the ACL token used for replicating intentions.
+	ConnectReplicationToken string
+
 	// ConnectTestDisableManagedProxies is not exposed to public config but us
 	// used by TestAgent to prevent self-executing the test binary in the
 	// background if a managed proxy is created for a test. The only place we
@@ -800,6 +803,13 @@ type RuntimeConfig struct {
 	// hcl: pid_file = string
 	PidFile string
 
+	// PrimaryDatacenter is the central datacenter that holds authoritative
+	// ACL records, replicates intentions and holds the root CA for Connect.
+	// This must be the same for the entire cluster. Off by default.
+	//
+	// hcl: primary_datacenter = string
+	PrimaryDatacenter string
+
 	// RPCAdvertiseAddr is the TCP address Consul advertises for its RPC endpoint.
 	// By default this is the bind address on the default RPC Server port. If the
 	// advertise address is specified then it is used.

agent/config/runtime_test.go

@@ -1379,7 +1379,9 @@ func TestConfigFlagsAndEdgecases(t *testing.T) {
 			patch: func(rt *RuntimeConfig) {
 				rt.ACLDatacenter = "a"
 				rt.DataDir = dataDir
+				rt.PrimaryDatacenter = "a"
 			},
+			warns: []string{`The 'acl_datacenter' field is deprecated. Use the 'primary_datacenter' field instead.`},
 		},
 		{
 			desc: "acl_replication_token enables acl replication",
@@ -1472,9 +1474,10 @@ func TestConfigFlagsAndEdgecases(t *testing.T) {
 				`-datacenter=a`,
 				`-data-dir=` + dataDir,
 			},
-			json: []string{`{ "acl_datacenter": "%" }`},
-			hcl:  []string{`acl_datacenter = "%"`},
-			err:  `acl_datacenter cannot be "%". Please use only [a-z0-9-_]`,
+			json:  []string{`{ "acl_datacenter": "%" }`},
+			hcl:   []string{`acl_datacenter = "%"`},
+			err:   `acl_datacenter cannot be "%". Please use only [a-z0-9-_]`,
+			warns: []string{`The 'acl_datacenter' field is deprecated. Use the 'primary_datacenter' field instead.`},
 		},
 		{
 			desc: "autopilot.max_trailing_logs invalid",
@@ -3013,6 +3016,7 @@ func TestFullConfig(t *testing.T) {
 				"sidecar_max_port": 9999
 			},
 			"protocol": 30793,
+			"primary_datacenter": "ejtmd43d",
 			"raft_protocol": 19016,
 			"raft_snapshot_threshold": 16384,
 			"raft_snapshot_interval": "30s",
@@ -3543,6 +3547,7 @@ func TestFullConfig(t *testing.T) {
 				sidecar_max_port = 9999
 			}
 			protocol = 30793
+			primary_datacenter = "ejtmd43d"
 			raft_protocol = 19016
 			raft_snapshot_threshold = 16384
 			raft_snapshot_interval = "30s"
@@ -4146,6 +4151,7 @@ func TestFullConfig(t *testing.T) {
 		NodeName:                         "otlLxGaI",
 		NonVotingServer:                  true,
 		PidFile:                          "43xN80Km",
+		PrimaryDatacenter:                "ejtmd43d",
 		RPCAdvertiseAddr:                 tcpAddr("17.99.29.16:3757"),
 		RPCBindAddr:                      tcpAddr("16.99.34.17:3757"),
 		RPCHoldTimeout:                   15707 * time.Second,
@@ -4488,6 +4494,7 @@ func TestFullConfig(t *testing.T) {
 	}
 
 	warns := []string{
+		`The 'acl_datacenter' field is deprecated. Use the 'primary_datacenter' field instead.`,
 		`bootstrap_expect > 0: expecting 53 servers`,
 	}
 
@@ -4849,6 +4856,7 @@ func TestSanitize(t *testing.T) {
 		"ConnectProxyDefaultScriptCommand": [],
 		"ConnectSidecarMaxPort": 0,
 		"ConnectSidecarMinPort": 0,
+		"ConnectReplicationToken": "hidden",
 		"ConnectTestDisableManagedProxies": false,
 		"ConsulCoordinateUpdateBatchSize": 0,
 		"ConsulCoordinateUpdateMaxBatches": 0,
@@ -4931,6 +4939,7 @@ func TestSanitize(t *testing.T) {
 		"NodeName": "",
 		"NonVotingServer": false,
 		"PidFile": "",
+		"PrimaryDatacenter": "",
 		"RPCAdvertiseAddr": "",
 		"RPCBindAddr": "",
 		"RPCHoldTimeout": "0s",

agent/consul/config.go

@@ -75,10 +75,14 @@ type Config struct {
 	// of nodes.
 	BootstrapExpect int
 
-	// Datacenter is the datacenter this Consul server represents
+	// Datacenter is the datacenter this Consul server represents.
 	Datacenter string
 
-	// DataDir is the directory to store our state in
+	// PrimaryDatacenter is the authoritative datacenter for features like ACLs
+	// and Connect.
+	PrimaryDatacenter string
+
+	// DataDir is the directory to store our state in.
 	DataDir string
 
 	// DevMode is used to enable a development server mode.
@@ -355,6 +359,9 @@ type Config struct {
 	// CAConfig is used to apply the initial Connect CA configuration when
 	// bootstrapping.
 	CAConfig *structs.CAConfiguration
+
+	// ConnectReplicationToken is used to control Intention replication.
+	ConnectReplicationToken string
 }
 
 // CheckProtocolVersion validates the protocol version.

agent/consul/connect_ca_endpoint.go

@@ -107,7 +107,7 @@ func (s *ConnectCA) ConfigurationSet(
 		return err
 	}
 
-	newActiveRoot, err := parseCARoot(newRootPEM, args.Config.Provider)
+	newActiveRoot, err := parseCARoot(newRootPEM, args.Config.Provider, args.Config.ClusterID)
 	if err != nil {
 		return err
 	}
@@ -120,7 +120,10 @@ func (s *ConnectCA) ConfigurationSet(
 		return err
 	}
 
-	if root != nil && root.ID == newActiveRoot.ID {
+	// If the root didn't change or if this is a secondary DC, just update the
+	// config and return.
+	if (s.srv.config.Datacenter != s.srv.config.PrimaryDatacenter) ||
+		root != nil && root.ID == newActiveRoot.ID {
 		args.Op = structs.CAOpSetConfig
 		resp, err := s.srv.raftApply(structs.ConnectCARequestType, args)
 		if err != nil {
@@ -276,16 +279,17 @@ func (s *ConnectCA) Roots(
 				// directly to the structure in the memdb store.
 
 				reply.Roots[i] = &structs.CARoot{
-					ID:                r.ID,
-					Name:              r.Name,
-					SerialNumber:      r.SerialNumber,
-					SigningKeyID:      r.SigningKeyID,
-					NotBefore:         r.NotBefore,
-					NotAfter:          r.NotAfter,
-					RootCert:          r.RootCert,
-					IntermediateCerts: r.IntermediateCerts,
-					RaftIndex:         r.RaftIndex,
-					Active:            r.Active,
+					ID:                  r.ID,
+					Name:                r.Name,
+					SerialNumber:        r.SerialNumber,
+					SigningKeyID:        r.SigningKeyID,
+					ExternalTrustDomain: r.ExternalTrustDomain,
+					NotBefore:           r.NotBefore,
+					NotAfter:            r.NotAfter,
+					RootCert:            r.RootCert,
+					IntermediateCerts:   r.IntermediateCerts,
+					RaftIndex:           r.RaftIndex,
+					Active:              r.Active,
 				}
 
 				if r.Active {

agent/consul/leader.go

@@ -227,6 +227,8 @@ func (s *Server) establishLeadership() error {
 		return err
 	}
 
+	s.startEnterpriseLeader()
+
 	s.startCARootPruning()
 
 	s.setConsistentReadReady()
@@ -245,6 +247,8 @@ func (s *Server) revokeLeadership() error {
 		return err
 	}
 
+	s.stopEnterpriseLeader()
+
 	s.stopCARootPruning()
 
 	s.setCAProvider(nil, nil)
@@ -414,24 +418,8 @@ func (s *Server) initializeCAConfig() (*structs.CAConfiguration, error) {
 	return config, nil
 }
 
-// initializeCA sets up the CA provider when gaining leadership, bootstrapping
-// the root in the state store if necessary.
-func (s *Server) initializeCA() error {
-	// Bail if connect isn't enabled.
-	if !s.config.ConnectEnabled {
-		return nil
-	}
-
-	conf, err := s.initializeCAConfig()
-	if err != nil {
-		return err
-	}
-
-	// Initialize the provider based on the current config.
-	provider, err := s.createCAProvider(conf)
-	if err != nil {
-		return err
-	}
+// initializeRootCA runs the initialization logic for a root CA.
+func (s *Server) initializeRootCA(provider ca.Provider, conf *structs.CAConfiguration) error {
 	if err := provider.Configure(conf.ClusterID, true, conf.Config); err != nil {
 		return fmt.Errorf("error configuring provider: %v", err)
 	}
@@ -445,7 +433,7 @@ func (s *Server) initializeCA() error {
 		return fmt.Errorf("error getting root cert: %v", err)
 	}
 
-	rootCA, err := parseCARoot(rootPEM, conf.Provider)
+	rootCA, err := parseCARoot(rootPEM, conf.Provider, conf.ClusterID)
 	if err != nil {
 		return err
 	}
@@ -495,13 +483,13 @@ func (s *Server) initializeCA() error {
 
 	s.setCAProvider(provider, rootCA)
 
-	s.logger.Printf("[INFO] connect: initialized CA with provider %q", conf.Provider)
+	s.logger.Printf("[INFO] connect: initialized primary datacenter CA with provider %q", conf.Provider)
 
 	return nil
 }
 
 // parseCARoot returns a filled-in structs.CARoot from a raw PEM value.
-func parseCARoot(pemValue, provider string) (*structs.CARoot, error) {
+func parseCARoot(pemValue, provider, clusterID string) (*structs.CARoot, error) {
 	id, err := connect.CalculateCertFingerprint(pemValue)
 	if err != nil {
 		return nil, fmt.Errorf("error parsing root fingerprint: %v", err)
@@ -511,14 +499,15 @@ func parseCARoot(pemValue, provider string) (*structs.CARoot, error) {
 		return nil, fmt.Errorf("error parsing root cert: %v", err)
 	}
 	return &structs.CARoot{
-		ID:           id,
-		Name:         fmt.Sprintf("%s CA Root Cert", strings.Title(provider)),
-		SerialNumber: rootCert.SerialNumber.Uint64(),
-		SigningKeyID: connect.HexString(rootCert.AuthorityKeyId),
-		NotBefore:    rootCert.NotBefore,
-		NotAfter:     rootCert.NotAfter,
-		RootCert:     pemValue,
-		Active:       true,
+		ID:                  id,
+		Name:                fmt.Sprintf("%s CA Root Cert", strings.Title(provider)),
+		SerialNumber:        rootCert.SerialNumber.Uint64(),
+		SigningKeyID:        connect.HexString(rootCert.AuthorityKeyId),
+		ExternalTrustDomain: clusterID,
+		NotBefore:           rootCert.NotBefore,
+		NotAfter:            rootCert.NotAfter,
+		RootCert:            pemValue,
+		Active:              true,
 	}, nil
 }
 

agent/consul/leader_oss.go

@@ -0,0 +1,29 @@
+// +build !ent
+
+package consul
+
+// initializeCA sets up the CA provider when gaining leadership, bootstrapping
+// the root in the state store if necessary.
+func (s *Server) initializeCA() error {
+	// Bail if connect isn't enabled.
+	if !s.config.ConnectEnabled {
+		return nil
+	}
+
+	conf, err := s.initializeCAConfig()
+	if err != nil {
+		return err
+	}
+
+	// Initialize the provider based on the current config.
+	provider, err := s.createCAProvider(conf)
+	if err != nil {
+		return err
+	}
+
+	return s.initializeRootCA(provider, conf)
+}
+
+// Stub methods, only present in Consul Enterprise.
+func (s *Server) startEnterpriseLeader() {}
+func (s *Server) stopEnterpriseLeader()  {}

agent/consul/server.go

@@ -274,6 +274,15 @@ func NewServerLogger(config *Config, logger *log.Logger, tokens *token.Store) (*
 		config.UseTLS = true
 	}
 
+	// Set the primary DC if it wasn't set.
+	if config.PrimaryDatacenter == "" {
+		if config.ACLDatacenter != "" {
+			config.PrimaryDatacenter = config.ACLDatacenter
+		} else {
+			config.PrimaryDatacenter = config.Datacenter
+		}
+	}
+
 	// Create the TLS wrapper for outgoing connections.
 	tlsConf := config.tlsConfig()
 	tlsWrap, err := tlsConf.OutgoingTLSWrapper()

agent/structs/connect_ca.go

@@ -54,6 +54,9 @@ type CARoot struct {
 	// private key used to sign the certificate.
 	SigningKeyID string
 
+	// ExternalTrustDomain is the trust domain this root was generated under.
+	ExternalTrustDomain string
+
 	// Time validity bounds.
 	NotBefore time.Time
 	NotAfter  time.Time