Merge pull request #4535 from hashicorp/ca-snapshot-fix

fsm: add missing CA config to snapshot/restore logic

Author: kyhavlov

agent/consul/fsm/fsm.go

@@ -14,7 +14,9 @@ import (
 )
 
 // msgpackHandle is a shared handle for encoding/decoding msgpack payloads
-var msgpackHandle = &codec.MsgpackHandle{}
+var msgpackHandle = &codec.MsgpackHandle{
+	RawToString: true,
+}
 
 // command is a command method on the FSM.
 type command func(buf []byte, index uint64) interface{}

agent/consul/fsm/snapshot_oss.go

@@ -23,6 +23,7 @@ func init() {
 	registerRestorer(structs.IntentionRequestType, restoreIntention)
 	registerRestorer(structs.ConnectCARequestType, restoreConnectCA)
 	registerRestorer(structs.ConnectCAProviderStateType, restoreConnectCAProviderState)
+	registerRestorer(structs.ConnectCAConfigType, restoreConnectCAConfig)
 }
 
 func persistOSS(s *snapshot, sink raft.SnapshotSink, encoder *codec.Encoder) error {
@@ -56,6 +57,9 @@ func persistOSS(s *snapshot, sink raft.SnapshotSink, encoder *codec.Encoder) err
 	if err := s.persistConnectCAProviderState(sink, encoder); err != nil {
 		return err
 	}
+	if err := s.persistConnectCAConfig(sink, encoder); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -286,6 +290,23 @@ func (s *snapshot) persistConnectCA(sink raft.SnapshotSink,
 			return err
 		}
 	}
+
+	return nil
+}
+
+func (s *snapshot) persistConnectCAConfig(sink raft.SnapshotSink,
+	encoder *codec.Encoder) error {
+	config, err := s.state.CAConfig()
+	if err != nil {
+		return err
+	}
+
+	if _, err := sink.Write([]byte{byte(structs.ConnectCAConfigType)}); err != nil {
+		return err
+	}
+	if err := encoder.Encode(config); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -464,3 +485,14 @@ func restoreConnectCAProviderState(header *snapshotHeader, restore *state.Restor
 	}
 	return nil
 }
+
+func restoreConnectCAConfig(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
+	var req structs.CAConfiguration
+	if err := decoder.Decode(&req); err != nil {
+		return err
+	}
+	if err := restore.CAConfig(&req); err != nil {
+		return err
+	}
+	return nil
+}

agent/consul/fsm/snapshot_oss_test.go

@@ -29,7 +29,27 @@ func TestFSM_SnapshotRestore_OSS(t *testing.T) {
 	// Add some state
 	fsm.state.EnsureNode(1, &structs.Node{Node: "foo", Address: "127.0.0.1"})
 	fsm.state.EnsureNode(2, &structs.Node{Node: "baz", Address: "127.0.0.2", TaggedAddresses: map[string]string{"hello": "1.2.3.4"}, Meta: map[string]string{"testMeta": "testing123"}})
-	fsm.state.EnsureService(3, "foo", &structs.NodeService{ID: "web", Service: "web", Tags: nil, Address: "127.0.0.1", Port: 80})
+
+	// Add a service instance with Connect config.
+	connectConf := structs.ServiceConnect{
+		Native: true,
+		Proxy: &structs.ServiceDefinitionConnectProxy{
+			Command:  []string{"foo", "bar"},
+			ExecMode: "a",
+			Config: map[string]interface{}{
+				"a": "qwer",
+				"b": 4.3,
+			},
+		},
+	}
+	fsm.state.EnsureService(3, "foo", &structs.NodeService{
+		ID:      "web",
+		Service: "web",
+		Tags:    nil,
+		Address: "127.0.0.1",
+		Port:    80,
+		Connect: connectConf,
+	})
 	fsm.state.EnsureService(4, "foo", &structs.NodeService{ID: "db", Service: "db", Tags: []string{"primary"}, Address: "127.0.0.1", Port: 5000})
 	fsm.state.EnsureService(5, "baz", &structs.NodeService{ID: "web", Service: "web", Tags: nil, Address: "127.0.0.2", Port: 80})
 	fsm.state.EnsureService(6, "baz", &structs.NodeService{ID: "db", Service: "db", Tags: []string{"secondary"}, Address: "127.0.0.2", Port: 5000})
@@ -131,6 +151,18 @@ func TestFSM_SnapshotRestore_OSS(t *testing.T) {
 	assert.Nil(err)
 	assert.True(ok)
 
+	// CA Config
+	caConfig := &structs.CAConfiguration{
+		ClusterID: "foo",
+		Provider:  "consul",
+		Config: map[string]interface{}{
+			"foo": "asdf",
+			"bar": 6.5,
+		},
+	}
+	err = fsm.state.CASetConfig(17, caConfig)
+	assert.Nil(err)
+
 	// Snapshot
 	snap, err := fsm.Snapshot()
 	if err != nil {
@@ -191,6 +223,10 @@ func TestFSM_SnapshotRestore_OSS(t *testing.T) {
 	if fooSrv.Services["db"].Port != 5000 {
 		t.Fatalf("Bad: %v", fooSrv)
 	}
+	connectSrv := fooSrv.Services["web"]
+	if !reflect.DeepEqual(connectSrv.Connect, connectConf) {
+		t.Fatalf("got: %v, want: %v", connectSrv.Connect, connectConf)
+	}
 
 	_, checks, err := fsm2.state.NodeChecks(nil, "foo")
 	if err != nil {
@@ -312,6 +348,11 @@ func TestFSM_SnapshotRestore_OSS(t *testing.T) {
 	assert.Equal("foo", state.PrivateKey)
 	assert.Equal("bar", state.RootCert)
 
+	// Verify CA configuration is restored.
+	_, caConf, err := fsm2.state.CAConfig()
+	assert.Nil(err)
+	assert.Equal(caConfig, caConf)
+
 	// Snapshot
 	snap, err = fsm2.Snapshot()
 	if err != nil {

agent/structs/structs.go

@@ -46,6 +46,7 @@ const (
 	IntentionRequestType                   = 12
 	ConnectCARequestType                   = 13
 	ConnectCAProviderStateType             = 14
+	ConnectCAConfigType                    = 15 // FSM snapshots only.
 )
 
 const (