SDS Server
多集群证书管理
初始化SDS Server
s.initSDSServer(args)func (s *Server) initSDSServer(args *PilotArgs) {
if s.kubeClient != nil {
// 是否校验客户端身份
if !features.EnableXDSIdentityCheck {
// Make sure we have security
log.Warnf("skipping Kubernetes credential reader; PILOT_ENABLE_XDS_IDENTITY_CHECK must be set to true for this feature.")
} else {
// 多集群初始化
sc := kubesecrets.NewMulticluster(s.kubeClient, s.clusterID, args.RegistryOptions.ClusterRegistriesNamespace, make(chan struct{}))
// 添加事件处理器，当有变化则推送XDS
sc.AddEventHandler(func(name, namespace string) {
s.XDSServer.ConfigUpdate(&model.PushRequest{
Full: false,
ConfigsUpdated: map[model.ConfigKey]struct{}{
{
Kind: gvk.Secret,
Name: name,
Namespace: namespace,
}: {},
},
Reason: []model.TriggerReason{model.SecretTrigger},
})
})
// sds Generator
s.XDSServer.Generators[v3.SecretType] = xds.NewSecretGen(sc, s.XDSServer.Cache)
}
}
}func NewMulticluster(client kube.Client, localCluster, secretNamespace string, stop chan struct{}) *Multicluster {
// 初始化
m := &Multicluster{
remoteKubeControllers: map[string]*SecretsController{},
localCluster: localCluster,
stop: stop,
}
// 添加本地集群
m.addMemberCluster(client, localCluster)
// 启动informer
sc := secretcontroller.StartSecretController(client,
func(c kube.Client, k string) error { m.addMemberCluster(c, k); return nil },
func(c kube.Client, k string) error { m.updateMemberCluster(c, k); return nil },
func(k string) error { m.deleteMemberCluster(k); return nil },
secretNamespace,
time.Millisecond*100,
stop)
m.secretController = sc
return m
}func NewController(
kubeclientset kubernetes.Interface,
namespace string,
cs *ClusterStore,
addCallback addSecretCallback,
updateCallback updateSecretCallback,
removeCallback removeSecretCallback) *Controller {
//获取informer 需要有 "istio/multiCluster=true"标签
secretsInformer := cache.NewSharedIndexInformer(
&cache.ListWatch{
ListFunc: func(opts meta_v1.ListOptions) (runtime.Object, error) {
opts.LabelSelector = MultiClusterSecretLabel + "=true"
return kubeclientset.CoreV1().Secrets(namespace).List(context.TODO(), opts)
},
WatchFunc: func(opts meta_v1.ListOptions) (watch.Interface, error) {
opts.LabelSelector = MultiClusterSecretLabel + "=true"
return kubeclientset.CoreV1().Secrets(namespace).Watch(context.TODO(), opts)
},
},
&corev1.Secret{}, 0, cache.Indexers{},
)
// 获取一个queue
queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
controller := &Controller{
kubeclientset: kubeclientset,
namespace: namespace,
cs: cs,
informer: secretsInformer,
queue: queue,
addCallback: addCallback,
updateCallback: updateCallback,
removeCallback: removeCallback,
}
log.Info("Setting up event handlers")
// 添加事件处理，全部放入queue
secretsInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
key, err := cache.MetaNamespaceKeyFunc(obj)
log.Infof("Processing add: %s", key)
if err == nil {
queue.Add(key)
}
},
UpdateFunc: func(oldObj, newObj interface{}) {
if oldObj == newObj || reflect.DeepEqual(oldObj, newObj) {
return
}
key, err := cache.MetaNamespaceKeyFunc(newObj)
log.Infof("Processing update: %s", key)
if err == nil {
queue.Add(key)
}
},
DeleteFunc: func(obj interface{}) {
key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
log.Infof("Processing delete: %s", key)
if err == nil {
queue.Add(key)
}
},
})
return controller
}事件处理
func (c *Controller) processNextItem() bool {
// 获取secret
secretName, quit := c.queue.Get()
if quit {
return false
}
defer c.queue.Done(secretName)
// 具体处理逻辑
err := c.processItem(secretName.(string))
if err == nil {
// No error, reset the ratelimit counters
c.queue.Forget(secretName)
} else if c.queue.NumRequeues(secretName) < maxRetries {
log.Errorf("Error processing %s (will retry): %v", secretName, err)
c.queue.AddRateLimited(secretName)
} else {
log.Errorf("Error processing %s (giving up): %v", secretName, err)
c.queue.Forget(secretName)
utilruntime.HandleError(err)
}
return true
}func (c *Controller) processItem(secretName string) error {
if secretName == initialSyncSignal {
c.initialSync.Store(true)
return nil
}
obj, exists, err := c.informer.GetIndexer().GetByKey(secretName)
if err != nil {
return fmt.Errorf("error fetching object %s error: %v", secretName, err)
}
if exists {
// 存在用addMemberCluster进行处理
c.addMemberCluster(secretName, obj.(*corev1.Secret))
} else {
// 不存在则删除
c.deleteMemberCluster(secretName)
}
return nil
}func (c *Controller) addMemberCluster(secretName string, s *corev1.Secret) {
for clusterID, kubeConfig := range s.Data {
// 判断cluster是否存在
if prev, ok := c.cs.remoteClusters[clusterID]; !ok {
log.Infof("Adding cluster_id=%v from secret=%v", clusterID, secretName)
// 创建remote集群对象
remoteCluster, err := createRemoteCluster(kubeConfig, secretName)
if err != nil {
log.Errorf("Failed to add remote cluster from secret=%v for cluster_id=%v: %v",
secretName, clusterID, err)
continue
}
// 写入remoteClusters
c.cs.remoteClusters[clusterID] = remoteCluster
// 调用addCallback
if err := c.addCallback(remoteCluster.clients, clusterID); err != nil {
log.Errorf("Error creating cluster_id=%s from secret %v: %v",
clusterID, secretName, err)
}
} else {
if prev.secretName != secretName {
log.Errorf("ClusterID reused in two different secrets: %v and %v. ClusterID "+
"must be unique across all secrets", prev.secretName, secretName)
continue
}
kubeConfigSha := sha256.Sum256(kubeConfig)
if bytes.Equal(kubeConfigSha[:], prev.kubeConfigSha[:]) {
log.Infof("Updating cluster_id=%v from secret=%v: (kubeconfig are identical)", clusterID, secretName)
} else {
log.Infof("Updating cluster %v from secret %v", clusterID, secretName)
remoteCluster, err := createRemoteCluster(kubeConfig, secretName)
if err != nil {
log.Errorf("Error updating cluster_id=%v from secret=%v: %v",
clusterID, secretName, err)
continue
}
c.cs.remoteClusters[clusterID] = remoteCluster
// 调用updateCallback
if err := c.updateCallback(remoteCluster.clients, clusterID); err != nil {
log.Errorf("Error updating cluster_id from secret=%v: %s %v",
clusterID, secretName, err)
}
}
}
}
log.Infof("Number of remote clusters: %d", len(c.cs.remoteClusters))
}func (c *Controller) deleteMemberCluster(secretName string) {
for clusterID, cluster := range c.cs.remoteClusters {
if cluster.secretName == secretName {
log.Infof("Deleting cluster_id=%v configured by secret=%v", clusterID, secretName)
// 调用remove call back
err := c.removeCallback(clusterID)
if err != nil {
log.Errorf("Error removing cluster_id=%v configured by secret=%v: %v",
clusterID, secretName, err)
}
delete(c.cs.remoteClusters, clusterID)
}
}
log.Infof("Number of remote clusters: %d", len(c.cs.remoteClusters))
}callbacks
addcallback
func (m *Multicluster) addMemberCluster(clients kube.Client, key string) {
log.Infof("initializing Kubernetes credential reader for cluster %v", key)
sc := NewSecretsController(clients, key)
m.m.Lock()
m.remoteKubeControllers[key] = sc
m.m.Unlock()
clients.RunAndWait(m.stop)
}updatecallback
func (m *Multicluster) updateMemberCluster(clients kube.Client, key string) {
m.deleteMemberCluster(key)
m.addMemberCluster(clients, key)
}delete callback
func (m *Multicluster) deleteMemberCluster(key string) {
m.m.Lock()
delete(m.remoteKubeControllers, key)
m.m.Unlock()
}Generate
func (s *SecretGen) Generate(proxy *model.Proxy, _ *model.PushContext, w *model.WatchedResource, req *model.PushRequest) model.Resources {
if proxy.VerifiedIdentity == nil {
adsLog.Warnf("proxy %v is not authorized to receive secrets. Ensure you are connecting over TLS port and are authenticated.", proxy.ID)
return nil
}
// 获取远程和本地集群的secretcontroller
secrets, err := s.secrets.ForCluster(proxy.Metadata.ClusterID)
if err != nil {
adsLog.Warnf("proxy %v is from an unknown cluster, cannot retrieve certificates: %v", proxy.ID, err)
return nil
}
// 认证客户端身份
if err := secrets.Authorize(proxy.VerifiedIdentity.ServiceAccount, proxy.VerifiedIdentity.Namespace); err != nil {
adsLog.Warnf("proxy %v is not authorized to receive secrets: %v", proxy.ID, err)
return nil
}
if req == nil || !needsUpdate(proxy, req.ConfigsUpdated) {
return nil
}
var updatedSecrets map[model.ConfigKey]struct{}
if !req.Full {
updatedSecrets = model.ConfigsOfKind(req.ConfigsUpdated, gvk.Secret)
}
results := model.Resources{}
for _, resource := range w.ResourceNames {
// 获取对应的secretresource
sr, err := parseResourceName(resource, proxy.ConfigNamespace)
if err != nil {
adsLog.Warnf("error parsing resource name: %v", err)
continue
}
if updatedSecrets != nil {
if !containsAny(updatedSecrets, relatedConfigs(model.ConfigKey{Kind: gvk.Secret, Name: sr.Name, Namespace: sr.Namespace})) {
// This is an incremental update, filter out secrets that are not updated.
continue
}
}
// 请求的资源必须和proxy同一命名空间
if err := s.proxyAuthorizedForSecret(proxy, sr); err != nil {
adsLog.Warnf("requested secret %v not accessible for proxy %v: %v", sr.ResourceName, proxy.ID, err)
continue
}
// 如果已经缓存则直接获取
if cached, f := s.cache.Get(sr); f {
// If it is in the Cache, add it and continue
results = append(results, cached)
continue
}
// 判断是否为网关根证书
isCAOnlySecret := strings.HasSuffix(sr.Name, GatewaySdsCaSuffix)
if isCAOnlySecret {
// 获取根证书
secret := secrets.GetCaCert(sr.Name, sr.Namespace)
if secret != nil {
res := toEnvoyCaSecret(sr.ResourceName, secret)
results = append(results, res)
s.cache.Add(sr, res)
} else {
adsLog.Warnf("failed to fetch ca certificate for %v", sr.ResourceName)
}
} else {
// 获取私钥和证书
key, cert := secrets.GetKeyAndCert(sr.Name, sr.Namespace)
if key != nil && cert != nil {
res := toEnvoyKeyCertSecret(sr.ResourceName, key, cert)
results = append(results, res)
s.cache.Add(sr, res)
} else {
adsLog.Warnf("failed to fetch key and certificate for %v", sr.ResourceName)
}
}
}
return results
}Last updated