diff --git a/agent/connect/ca.go b/agent/connect/ca.go
index bca9392d3..818af9f9f 100644
--- a/agent/connect/ca.go
+++ b/agent/connect/ca.go
@@ -38,6 +38,18 @@ func ParseSigner(pemValue string) (crypto.Signer, error) {
 	case "EC PRIVATE KEY":
 		return x509.ParseECPrivateKey(block.Bytes)
 
+	case "PRIVATE KEY":
+		signer, err := x509.ParsePKCS8PrivateKey(block.Bytes)
+		if err != nil {
+			return nil, err
+		}
+		pk, ok := signer.(crypto.Signer)
+		if !ok {
+			return nil, fmt.Errorf("private key is not a valid format")
+		}
+
+		return pk, nil
+
 	default:
 		return nil, fmt.Errorf("unknown PEM block type for signing key: %s", block.Type)
 	}
diff --git a/agent/connect/ca_provider.go b/agent/connect/ca_provider.go
index ca0ccf9b0..dc70c6a58 100644
--- a/agent/connect/ca_provider.go
+++ b/agent/connect/ca_provider.go
@@ -10,9 +10,10 @@ import (
 // an external CA that provides leaf certificate signing for
 // given SpiffeIDServices.
 type CAProvider interface {
-	SetConfiguration(raw map[string]interface{}) error
 	ActiveRoot() (*structs.CARoot, error)
 	ActiveIntermediate() (*structs.CARoot, error)
 	GenerateIntermediate() (*structs.CARoot, error)
 	Sign(*SpiffeIDService, *x509.CertificateRequest) (*structs.IssuedCert, error)
+	//SignCA(*x509.CertificateRequest) (*structs.IssuedCert, error)
+	Teardown() error
 }
diff --git a/agent/consul/connect_ca_endpoint.go b/agent/consul/connect_ca_endpoint.go
index d0c582165..128c1493d 100644
--- a/agent/consul/connect_ca_endpoint.go
+++ b/agent/consul/connect_ca_endpoint.go
@@ -2,6 +2,7 @@ package consul
 
 import (
 	"fmt"
+	"reflect"
 
 	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/agent/connect"
@@ -60,9 +61,95 @@ func (s *ConnectCA) ConfigurationSet(
 		return acl.ErrPermissionDenied
 	}
 
-	// Commit
-	// todo(kyhavlov): trigger a bootstrap here when the provider changes
-	args.Op = structs.CAOpSetConfig
+	// Exit early if it's a no-op change
+	state := s.srv.fsm.State()
+	_, config, err := state.CAConfig()
+	if err != nil {
+		return err
+	}
+	if args.Config.Provider == config.Provider && reflect.DeepEqual(args.Config.Config, config.Config) {
+		return nil
+	}
+
+	// Create a new instance of the provider described by the config
+	// and get the current active root CA. This acts as a good validation
+	// of the config and makes sure the provider is functioning correctly
+	// before we commit any changes to Raft.
+	newProvider, err := s.srv.createCAProvider(args.Config)
+	if err != nil {
+		return fmt.Errorf("could not initialize provider: %v", err)
+	}
+
+	newActiveRoot, err := newProvider.ActiveRoot()
+	if err != nil {
+		return err
+	}
+
+	// Compare the new provider's root CA ID to the current one. If they
+	// match, just update the existing provider with the new config.
+	// If they don't match, begin the root rotation process.
+	_, root, err := state.CARootActive(nil)
+	if err != nil {
+		return err
+	}
+
+	if root != nil && root.ID == newActiveRoot.ID {
+		args.Op = structs.CAOpSetConfig
+		resp, err := s.srv.raftApply(structs.ConnectCARequestType, args)
+		if err != nil {
+			return err
+		}
+		if respErr, ok := resp.(error); ok {
+			return respErr
+		}
+
+		// If the config has been committed, update the local provider instance
+		s.srv.setCAProvider(newProvider)
+
+		s.srv.logger.Printf("[INFO] connect: provider config updated")
+
+		return nil
+	}
+
+	// At this point, we know the config change has trigged a root rotation,
+	// either by swapping the provider type or changing the provider's config
+	// to use a different root certificate.
+
+	// If it's a config change that would trigger a rotation (different provider/root):
+	// -1. Create an instance of the provider described by the new config
+	// 2. Get the intermediate from the new provider
+	// 3. Generate a CSR for the new intermediate, call SignCA on the old/current provider
+	// to get the cross-signed intermediate
+	// ~4. Get the active root for the new provider, append the intermediate from step 3
+	// to its list of intermediates
+	// -5. Update the roots and CA config in the state store at the same time, finally switching
+	// to the new provider
+	// -6. Call teardown on the old provider, so it can clean up whatever it needs to
+
+	/*_, err := newProvider.ActiveIntermediate()
+	if err != nil {
+		return err
+	}*/
+
+	// Update the roots and CA config in the state store at the same time
+	idx, roots, err := state.CARoots(nil)
+	if err != nil {
+		return err
+	}
+
+	var newRoots structs.CARoots
+	for _, r := range roots {
+		newRoot := *r
+		if newRoot.Active {
+			newRoot.Active = false
+		}
+		newRoots = append(newRoots, &newRoot)
+	}
+	newRoots = append(newRoots, newActiveRoot)
+
+	args.Op = structs.CAOpSetRootsAndConfig
+	args.Index = idx
+	args.Roots = newRoots
 	resp, err := s.srv.raftApply(structs.ConnectCARequestType, args)
 	if err != nil {
 		return err
@@ -71,6 +158,17 @@ func (s *ConnectCA) ConfigurationSet(
 		return respErr
 	}
 
+	// If the config has been committed, update the local provider instance
+	// and call teardown on the old provider
+	oldProvider := s.srv.getCAProvider()
+	s.srv.setCAProvider(newProvider)
+
+	if err := oldProvider.Teardown(); err != nil {
+		return err
+	}
+
+	s.srv.logger.Printf("[INFO] connect: CA rotated to the new root under %q provider", args.Config.Provider)
+
 	return nil
 }
 
diff --git a/agent/consul/connect_ca_provider.go b/agent/consul/connect_ca_provider.go
index 9beb6bfac..b72a9ee36 100644
--- a/agent/consul/connect_ca_provider.go
+++ b/agent/consul/connect_ca_provider.go
@@ -29,28 +29,94 @@ type ConsulCAProviderConfig struct {
 type ConsulCAProvider struct {
 	config *ConsulCAProviderConfig
 
-	// todo(kyhavlov): store these directly in the state store
-	// and pass a reference to the state to this provider instead of
-	// having these values here
+	id  string
 	srv *Server
 	sync.RWMutex
 }
 
+// NewConsulCAProvider returns a new instance of the Consul CA provider,
+// bootstrapping its state in the state store necessary
 func NewConsulCAProvider(rawConfig map[string]interface{}, srv *Server) (*ConsulCAProvider, error) {
-	provider := &ConsulCAProvider{srv: srv}
-	provider.SetConfiguration(rawConfig)
-
-	return provider, nil
-}
-
-func (c *ConsulCAProvider) SetConfiguration(raw map[string]interface{}) error {
-	conf, err := decodeConfig(raw)
+	conf, err := decodeConfig(rawConfig)
 	if err != nil {
-		return err
+		return nil, err
+	}
+	provider := &ConsulCAProvider{
+		config: conf,
+		srv:    srv,
+		id:     fmt.Sprintf("%s,%s", conf.PrivateKey, conf.RootCert),
 	}
 
-	c.config = conf
-	return nil
+	// Check if this configuration of the provider has already been
+	// initialized in the state store.
+	state := srv.fsm.State()
+	_, providerState, err := state.CAProviderState(provider.id)
+	if err != nil {
+		return nil, err
+	}
+
+	// Exit early if the state store has already been populated for this config.
+	if providerState != nil {
+		return provider, nil
+	}
+
+	newState := structs.CAConsulProviderState{
+		ID: provider.id,
+	}
+
+	// Write the initial provider state to get the index to use for the
+	// CA serial number.
+	{
+		args := &structs.CARequest{
+			Op:            structs.CAOpSetProviderState,
+			ProviderState: &newState,
+		}
+		resp, err := srv.raftApply(structs.ConnectCARequestType, args)
+		if err != nil {
+			return nil, err
+		}
+		if respErr, ok := resp.(error); ok {
+			return nil, respErr
+		}
+	}
+
+	idx, _, err := state.CAProviderState(provider.id)
+	if err != nil {
+		return nil, err
+	}
+
+	// Generate a private key if needed
+	if conf.PrivateKey == "" {
+		pk, err := generatePrivateKey()
+		if err != nil {
+			return nil, err
+		}
+		newState.PrivateKey = pk
+	} else {
+		newState.PrivateKey = conf.PrivateKey
+	}
+
+	// Generate the root CA
+	ca, err := provider.generateCA(newState.PrivateKey, conf.RootCert, idx+1)
+	if err != nil {
+		return nil, fmt.Errorf("error generating CA: %v", err)
+	}
+	newState.CARoot = ca
+
+	// Write the provider state
+	args := &structs.CARequest{
+		Op:            structs.CAOpSetProviderState,
+		ProviderState: &newState,
+	}
+	resp, err := srv.raftApply(structs.ConnectCARequestType, args)
+	if err != nil {
+		return nil, err
+	}
+	if respErr, ok := resp.(error); ok {
+		return nil, respErr
+	}
+
+	return provider, nil
 }
 
 func decodeConfig(raw map[string]interface{}) (*ConsulCAProviderConfig, error) {
@@ -59,59 +125,22 @@ func decodeConfig(raw map[string]interface{}) (*ConsulCAProviderConfig, error) {
 		return nil, fmt.Errorf("error decoding config: %s", err)
 	}
 
+	if config.PrivateKey == "" && config.RootCert != "" {
+		return nil, fmt.Errorf("must provide a private key when providing a root cert")
+	}
+
 	return config, nil
 }
 
 // Return the active root CA and generate a new one if needed
 func (c *ConsulCAProvider) ActiveRoot() (*structs.CARoot, error) {
 	state := c.srv.fsm.State()
-	_, providerState, err := state.CAProviderState()
+	_, providerState, err := state.CAProviderState(c.id)
 	if err != nil {
 		return nil, err
 	}
 
-	var update bool
-	var newState structs.CAConsulProviderState
-	if providerState != nil {
-		newState = *providerState
-	}
-
-	// Generate a private key if needed
-	if providerState == nil || providerState.PrivateKey == "" {
-		pk, err := generatePrivateKey()
-		if err != nil {
-			return nil, err
-		}
-		newState.PrivateKey = pk
-		update = true
-	}
-
-	// Generate a root CA if needed
-	if providerState == nil || providerState.CARoot == nil {
-		ca, err := c.generateCA(newState.PrivateKey, newState.RootIndex+1)
-		if err != nil {
-			return nil, err
-		}
-		newState.CARoot = ca
-		newState.RootIndex += 1
-		update = true
-	}
-
-	// Update the provider state if we generated a new private key/cert
-	if update {
-		args := &structs.CARequest{
-			Op:            structs.CAOpSetProviderState,
-			ProviderState: &newState,
-		}
-		resp, err := c.srv.raftApply(structs.ConnectCARequestType, args)
-		if err != nil {
-			return nil, err
-		}
-		if respErr, ok := resp.(error); ok {
-			return nil, respErr
-		}
-	}
-	return newState.CARoot, nil
+	return providerState.CARoot, nil
 }
 
 func (c *ConsulCAProvider) ActiveIntermediate() (*structs.CARoot, error) {
@@ -120,15 +149,12 @@ func (c *ConsulCAProvider) ActiveIntermediate() (*structs.CARoot, error) {
 
 func (c *ConsulCAProvider) GenerateIntermediate() (*structs.CARoot, error) {
 	state := c.srv.fsm.State()
-	_, providerState, err := state.CAProviderState()
+	idx, providerState, err := state.CAProviderState(c.id)
 	if err != nil {
 		return nil, err
 	}
-	if providerState == nil {
-		return nil, fmt.Errorf("CA provider not yet initialized")
-	}
 
-	ca, err := c.generateCA(providerState.PrivateKey, providerState.RootIndex+1)
+	ca, err := c.generateCA(providerState.PrivateKey, "", idx+1)
 	if err != nil {
 		return nil, err
 	}
@@ -136,12 +162,34 @@ func (c *ConsulCAProvider) GenerateIntermediate() (*structs.CARoot, error) {
 	return ca, nil
 }
 
+// Remove the state store entry for this provider instance.
+func (c *ConsulCAProvider) Teardown() error {
+	args := &structs.CARequest{
+		Op:            structs.CAOpDeleteProviderState,
+		ProviderState: &structs.CAConsulProviderState{ID: c.id},
+	}
+	resp, err := c.srv.raftApply(structs.ConnectCARequestType, args)
+	if err != nil {
+		return err
+	}
+	if respErr, ok := resp.(error); ok {
+		return respErr
+	}
+
+	return nil
+}
+
 // Sign returns a new certificate valid for the given SpiffeIDService
 // using the current CA.
 func (c *ConsulCAProvider) Sign(serviceId *connect.SpiffeIDService, csr *x509.CertificateRequest) (*structs.IssuedCert, error) {
+	// Lock during the signing so we don't use the same index twice
+	// for different cert serial numbers.
+	c.Lock()
+	defer c.Unlock()
+
 	// Get the provider state
 	state := c.srv.fsm.State()
-	_, providerState, err := state.CAProviderState()
+	_, providerState, err := state.CAProviderState(c.id)
 	if err != nil {
 		return nil, err
 	}
@@ -254,7 +302,7 @@ func generatePrivateKey() (string, error) {
 }
 
 // generateCA makes a new root CA using the current private key
-func (c *ConsulCAProvider) generateCA(privateKey string, sn uint64) (*structs.CARoot, error) {
+func (c *ConsulCAProvider) generateCA(privateKey, contents string, sn uint64) (*structs.CARoot, error) {
 	state := c.srv.fsm.State()
 	_, config, err := state.CAConfig()
 	if err != nil {
@@ -263,48 +311,54 @@ func (c *ConsulCAProvider) generateCA(privateKey string, sn uint64) (*structs.CA
 
 	privKey, err := connect.ParseSigner(privateKey)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("error parsing private key %q: %v", privateKey, err)
 	}
 
 	name := fmt.Sprintf("Consul CA %d", sn)
 
-	// The URI (SPIFFE compatible) for the cert
-	id := &connect.SpiffeIDSigning{ClusterID: config.ClusterSerial, Domain: "consul"}
-	keyId, err := connect.KeyId(privKey.Public())
-	if err != nil {
-		return nil, err
-	}
+	pemContents := contents
 
-	// Create the CA cert
-	serialNum := &big.Int{}
-	serialNum.SetUint64(sn)
-	template := x509.Certificate{
-		SerialNumber: serialNum,
-		Subject:      pkix.Name{CommonName: name},
-		URIs:         []*url.URL{id.URI()},
-		PermittedDNSDomainsCritical: true,
-		PermittedDNSDomains:         []string{id.URI().Hostname()},
-		BasicConstraintsValid:       true,
-		KeyUsage: x509.KeyUsageCertSign |
-			x509.KeyUsageCRLSign |
-			x509.KeyUsageDigitalSignature,
-		IsCA:           true,
-		NotAfter:       time.Now().Add(10 * 365 * 24 * time.Hour),
-		NotBefore:      time.Now(),
-		AuthorityKeyId: keyId,
-		SubjectKeyId:   keyId,
-	}
+	if pemContents == "" {
+		// The URI (SPIFFE compatible) for the cert
+		id := &connect.SpiffeIDSigning{ClusterID: config.ClusterSerial, Domain: "consul"}
+		keyId, err := connect.KeyId(privKey.Public())
+		if err != nil {
+			return nil, err
+		}
 
-	bs, err := x509.CreateCertificate(
-		rand.Reader, &template, &template, privKey.Public(), privKey)
-	if err != nil {
-		return nil, fmt.Errorf("error generating CA certificate: %s", err)
-	}
+		// Create the CA cert
+		serialNum := &big.Int{}
+		serialNum.SetUint64(sn)
+		template := x509.Certificate{
+			SerialNumber: serialNum,
+			Subject:      pkix.Name{CommonName: name},
+			URIs:         []*url.URL{id.URI()},
+			PermittedDNSDomainsCritical: true,
+			PermittedDNSDomains:         []string{id.URI().Hostname()},
+			BasicConstraintsValid:       true,
+			KeyUsage: x509.KeyUsageCertSign |
+				x509.KeyUsageCRLSign |
+				x509.KeyUsageDigitalSignature,
+			IsCA:           true,
+			NotAfter:       time.Now().Add(10 * 365 * 24 * time.Hour),
+			NotBefore:      time.Now(),
+			AuthorityKeyId: keyId,
+			SubjectKeyId:   keyId,
+		}
 
-	var buf bytes.Buffer
-	err = pem.Encode(&buf, &pem.Block{Type: "CERTIFICATE", Bytes: bs})
-	if err != nil {
-		return nil, fmt.Errorf("error encoding private key: %s", err)
+		bs, err := x509.CreateCertificate(
+			rand.Reader, &template, &template, privKey.Public(), privKey)
+		if err != nil {
+			return nil, fmt.Errorf("error generating CA certificate: %s", err)
+		}
+
+		var buf bytes.Buffer
+		err = pem.Encode(&buf, &pem.Block{Type: "CERTIFICATE", Bytes: bs})
+		if err != nil {
+			return nil, fmt.Errorf("error encoding private key: %s", err)
+		}
+
+		pemContents = buf.String()
 	}
 
 	// Generate an ID for the new CA cert
@@ -316,7 +370,7 @@ func (c *ConsulCAProvider) generateCA(privateKey string, sn uint64) (*structs.CA
 	return &structs.CARoot{
 		ID:       rootId,
 		Name:     name,
-		RootCert: buf.String(),
+		RootCert: pemContents,
 		Active:   true,
 	}, nil
 }
diff --git a/agent/consul/fsm/commands_oss.go b/agent/consul/fsm/commands_oss.go
index 99755194b..5292bd0f5 100644
--- a/agent/consul/fsm/commands_oss.go
+++ b/agent/consul/fsm/commands_oss.go
@@ -307,6 +307,23 @@ func (c *FSM) applyConnectCAOperation(buf []byte, index uint64) interface{} {
 			return err
 		}
 
+		return act
+	case structs.CAOpDeleteProviderState:
+		if err := c.state.CADeleteProviderState(req.ProviderState.ID); err != nil {
+			return err
+		}
+
+		return true
+	case structs.CAOpSetRootsAndConfig:
+		act, err := c.state.CARootSetCAS(index, req.Index, req.Roots)
+		if err != nil {
+			return err
+		}
+
+		if err := c.state.CASetConfig(index+1, req.Config); err != nil {
+			return err
+		}
+
 		return act
 	default:
 		c.logger.Printf("[WARN] consul.fsm: Invalid CA operation '%s'", req.Op)
diff --git a/agent/consul/leader.go b/agent/consul/leader.go
index fca3fa07f..8d62ca1aa 100644
--- a/agent/consul/leader.go
+++ b/agent/consul/leader.go
@@ -396,7 +396,7 @@ func (s *Server) getOrCreateCAConfig() (*structs.CAConfiguration, error) {
 	return config, nil
 }
 
-// bootstrapCA handles the initialization of a new CA provider
+// bootstrapCA creates a CA provider from the current configuration.
 func (s *Server) bootstrapCA() error {
 	conf, err := s.getOrCreateCAConfig()
 	if err != nil {
@@ -404,20 +404,12 @@ func (s *Server) bootstrapCA() error {
 	}
 
 	// Initialize the right provider based on the config
-	var provider connect.CAProvider
-	switch conf.Provider {
-	case structs.ConsulCAProvider:
-		provider, err = NewConsulCAProvider(conf.Config, s)
-		if err != nil {
-			return err
-		}
-	default:
-		return fmt.Errorf("unknown CA provider %q", conf.Provider)
+	provider, err := s.createCAProvider(conf)
+	if err != nil {
+		return err
 	}
 
-	s.caProviderLock.Lock()
-	s.caProvider = provider
-	s.caProviderLock.Unlock()
+	s.setCAProvider(provider)
 
 	// Get the active root cert from the CA
 	trustedCA, err := provider.ActiveRoot()
@@ -425,13 +417,14 @@ func (s *Server) bootstrapCA() error {
 		return fmt.Errorf("error getting root cert: %v", err)
 	}
 
-	// Check if this CA is already initialized
+	// Check if the CA root is already initialized and exit if it is.
+	// Every change to the CA after this initial bootstrapping should
+	// be done through the rotation process.
 	state := s.fsm.State()
 	_, root, err := state.CARootActive(nil)
 	if err != nil {
 		return err
 	}
-	// Exit early if the root is already in the state store.
 	if root != nil && root.ID == trustedCA.ID {
 		return nil
 	}
@@ -461,6 +454,28 @@ func (s *Server) bootstrapCA() error {
 	return nil
 }
 
+// createProvider returns a connect CA provider from the given config.
+func (s *Server) createCAProvider(conf *structs.CAConfiguration) (connect.CAProvider, error) {
+	switch conf.Provider {
+	case structs.ConsulCAProvider:
+		return NewConsulCAProvider(conf.Config, s)
+	default:
+		return nil, fmt.Errorf("unknown CA provider %q", conf.Provider)
+	}
+}
+
+func (s *Server) getCAProvider() connect.CAProvider {
+	s.caProviderLock.RLock()
+	defer s.caProviderLock.RUnlock()
+	return s.caProvider
+}
+
+func (s *Server) setCAProvider(newProvider connect.CAProvider) {
+	s.caProviderLock.Lock()
+	defer s.caProviderLock.Unlock()
+	s.caProvider = newProvider
+}
+
 // signConnectCert signs a cert for a service using the currently configured CA provider
 func (s *Server) signConnectCert(service *connect.SpiffeIDService, csr *x509.CertificateRequest) (*structs.IssuedCert, error) {
 	s.caProviderLock.RLock()
diff --git a/agent/consul/state/connect_ca.go b/agent/consul/state/connect_ca.go
index 2cce8028b..17e274992 100644
--- a/agent/consul/state/connect_ca.go
+++ b/agent/consul/state/connect_ca.go
@@ -60,8 +60,8 @@ func caProviderTableSchema() *memdb.TableSchema {
 				Name:         "id",
 				AllowMissing: false,
 				Unique:       true,
-				Indexer: &memdb.ConditionalIndex{
-					Conditional: func(obj interface{}) (bool, error) { return true, nil },
+				Indexer: &memdb.StringFieldIndex{
+					Field: "ID",
 				},
 			},
 		},
@@ -98,12 +98,12 @@ func (s *Restore) CAConfig(config *structs.CAConfiguration) error {
 	return nil
 }
 
-// CAConfig is used to get the current Autopilot configuration.
+// CAConfig is used to get the current CA configuration.
 func (s *Store) CAConfig() (uint64, *structs.CAConfiguration, error) {
 	tx := s.db.Txn(false)
 	defer tx.Abort()
 
-	// Get the autopilot config
+	// Get the CA config
 	c, err := tx.First(caConfigTableName, "id")
 	if err != nil {
 		return 0, nil, fmt.Errorf("failed CA config lookup: %s", err)
@@ -117,7 +117,7 @@ func (s *Store) CAConfig() (uint64, *structs.CAConfiguration, error) {
 	return config.ModifyIndex, config, nil
 }
 
-// CASetConfig is used to set the current Autopilot configuration.
+// CASetConfig is used to set the current CA configuration.
 func (s *Store) CASetConfig(idx uint64, config *structs.CAConfiguration) error {
 	tx := s.db.Txn(true)
 	defer tx.Abort()
@@ -341,13 +341,16 @@ func (s *Restore) CAProviderState(state *structs.CAConsulProviderState) error {
 	return nil
 }
 
-// CAProviderState is used to get the current Consul CA provider state.
-func (s *Store) CAProviderState() (uint64, *structs.CAConsulProviderState, error) {
+// CAProviderState is used to get the Consul CA provider state for the given ID.
+func (s *Store) CAProviderState(id string) (uint64, *structs.CAConsulProviderState, error) {
 	tx := s.db.Txn(false)
 	defer tx.Abort()
 
-	// Get the autopilot config
-	c, err := tx.First(caProviderTableName, "id")
+	// Get the index
+	idx := maxIndexTxn(tx, caProviderTableName)
+
+	// Get the provider config
+	c, err := tx.First(caProviderTableName, "id", id)
 	if err != nil {
 		return 0, nil, fmt.Errorf("failed built-in CA state lookup: %s", err)
 	}
@@ -357,7 +360,28 @@ func (s *Store) CAProviderState() (uint64, *structs.CAConsulProviderState, error
 		return 0, nil, nil
 	}
 
-	return state.ModifyIndex, state, nil
+	return idx, state, nil
+}
+
+// CAProviderStates is used to get the Consul CA provider state for the given ID.
+func (s *Store) CAProviderStates() (uint64, []*structs.CAConsulProviderState, error) {
+	tx := s.db.Txn(false)
+	defer tx.Abort()
+
+	// Get the index
+	idx := maxIndexTxn(tx, caProviderTableName)
+
+	// Get all
+	iter, err := tx.Get(caProviderTableName, "id")
+	if err != nil {
+		return 0, nil, fmt.Errorf("failed CA provider state lookup: %s", err)
+	}
+
+	var results []*structs.CAConsulProviderState
+	for v := iter.Next(); v != nil; v = iter.Next() {
+		results = append(results, v.(*structs.CAConsulProviderState))
+	}
+	return idx, results, nil
 }
 
 // CASetProviderState is used to set the current built-in CA provider state.
@@ -366,14 +390,14 @@ func (s *Store) CASetProviderState(idx uint64, state *structs.CAConsulProviderSt
 	defer tx.Abort()
 
 	// Check for an existing config
-	existing, err := tx.First(caProviderTableName, "id")
+	existing, err := tx.First(caProviderTableName, "id", state.ID)
 	if err != nil {
 		return false, fmt.Errorf("failed built-in CA state lookup: %s", err)
 	}
 
 	// Set the indexes.
 	if existing != nil {
-		state.CreateIndex = existing.(*structs.CAConfiguration).CreateIndex
+		state.CreateIndex = existing.(*structs.CAConsulProviderState).CreateIndex
 	} else {
 		state.CreateIndex = idx
 	}
@@ -382,7 +406,45 @@ func (s *Store) CASetProviderState(idx uint64, state *structs.CAConsulProviderSt
 	if err := tx.Insert(caProviderTableName, state); err != nil {
 		return false, fmt.Errorf("failed updating built-in CA state: %s", err)
 	}
+
+	// Update the index
+	if err := tx.Insert("index", &IndexEntry{caProviderTableName, idx}); err != nil {
+		return false, fmt.Errorf("failed updating index: %s", err)
+	}
+
 	tx.Commit()
 
 	return true, nil
 }
+
+// CADeleteProviderState is used to remove the Consul CA provider state for the given ID.
+func (s *Store) CADeleteProviderState(id string) error {
+	tx := s.db.Txn(true)
+	defer tx.Abort()
+
+	// Get the index
+	idx := maxIndexTxn(tx, caProviderTableName)
+
+	// Check for an existing config
+	existing, err := tx.First(caProviderTableName, "id", id)
+	if err != nil {
+		return fmt.Errorf("failed built-in CA state lookup: %s", err)
+	}
+	if existing == nil {
+		return nil
+	}
+
+	providerState := existing.(*structs.CAConsulProviderState)
+
+	// Do the delete and update the index
+	if err := tx.Delete(caProviderTableName, providerState); err != nil {
+		return err
+	}
+	if err := tx.Insert("index", &IndexEntry{caProviderTableName, idx}); err != nil {
+		return fmt.Errorf("failed updating index: %s", err)
+	}
+
+	tx.Commit()
+
+	return nil
+}
diff --git a/agent/structs/connect_ca.go b/agent/structs/connect_ca.go
index a923c0361..1e2959dd1 100644
--- a/agent/structs/connect_ca.go
+++ b/agent/structs/connect_ca.go
@@ -96,9 +96,11 @@ type IssuedCert struct {
 type CAOp string
 
 const (
-	CAOpSetRoots         CAOp = "set-roots"
-	CAOpSetConfig        CAOp = "set-config"
-	CAOpSetProviderState CAOp = "set-provider-state"
+	CAOpSetRoots            CAOp = "set-roots"
+	CAOpSetConfig           CAOp = "set-config"
+	CAOpSetProviderState    CAOp = "set-provider-state"
+	CAOpDeleteProviderState CAOp = "delete-provider-state"
+	CAOpSetRootsAndConfig   CAOp = "set-roots-config"
 )
 
 // CARequest is used to modify connect CA data. This is used by the
@@ -156,9 +158,9 @@ type CAConfiguration struct {
 
 // CAConsulProviderState is used to track the built-in Consul CA provider's state.
 type CAConsulProviderState struct {
+	ID         string
 	PrivateKey string
 	CARoot     *CARoot
-	RootIndex  uint64
 	LeafIndex  uint64
 
 	RaftIndex