XDS Server
XDS Server
istiod 中 xds Server 分为Secure/insecure 两种
initSecureDiscoveryService
对于启用tls的DiscoveryService需要先初始化spiffe验证器,对应的方法为setPeerCertVerifier
setPeerCertVerifier
func (s *Server) setPeerCertVerifier(tlsOptions TLSOptions) error {
if tlsOptions.CaCertFile == "" && s.CA == nil && features.SpiffeBundleEndpoints == "" {
// Running locally without configured certs - no TLS mode
return nil
}
s.peerCertVerifier = spiffe.NewPeerCertVerifier()
var rootCertBytes []byte
var err error
// 判断是否手动指定
if tlsOptions.CaCertFile != "" {
if rootCertBytes, err = ioutil.ReadFile(tlsOptions.CaCertFile); err != nil {
return err
}
} else {
// 加载RA cert
if s.RA != nil {
rootCertBytes = append(rootCertBytes, s.RA.GetCAKeyCertBundle().GetRootCertPem()...)
}
// 加载CA cert
if s.CA != nil {
rootCertBytes = append(rootCertBytes, s.CA.GetCAKeyCertBundle().GetRootCertPem()...)
}
}
if len(rootCertBytes) != 0 {
// 根据信任域添加添加CA证书到certPools/generalCertPool
err := s.peerCertVerifier.AddMappingFromPEM(spiffe.GetTrustDomain(), rootCertBytes)
if err != nil {
log.Errorf("Add Root CAs into peerCertVerifier failed: %v", err)
return fmt.Errorf("add root CAs into peerCertVerifier failed: %v", err)
}
}
if features.SpiffeBundleEndpoints != "" {
certMap, err := spiffe.RetrieveSpiffeBundleRootCertsFromStringInput(
features.SpiffeBundleEndpoints, []*x509.Certificate{})
if err != nil {
return err
}
s.peerCertVerifier.AddMappings(certMap)
}
return nil
}在initSecureDiscoveryService初始化grpc server时通过该验证器进行验证客户端身份
cfg := &tls.Config{
GetCertificate: s.getIstiodCertificate,
ClientAuth: tls.VerifyClientCertIfGiven,
ClientCAs: s.peerCertVerifier.GetGeneralCertPool(),
VerifyPeerCertificate: func(rawCerts [][]byte, verifiedChains [][]*x509.Certificate) error {
err := s.peerCertVerifier.VerifyPeerCert(rawCerts, verifiedChains)
if err != nil {
log.Infof("Could not verify certificate: %v", err)
}
return err
},
}具体验证客户端cert的逻辑如下
func (v *PeerCertVerifier) VerifyPeerCert(rawCerts [][]byte, _ [][]*x509.Certificate) error {
if len(rawCerts) == 0 {
// Peer doesn't present a certificate. Just skip. Other authn methods may be used.
return nil
}
var peerCert *x509.Certificate
intCertPool := x509.NewCertPool()
for id, rawCert := range rawCerts {
cert, err := x509.ParseCertificate(rawCert)
if err != nil {
return err
}
if id == 0 {
peerCert = cert
} else {
intCertPool.AddCert(cert)
}
}
if len(peerCert.URIs) != 1 {
return fmt.Errorf("peer certificate does not contain 1 URI type SAN, detected %d", len(peerCert.URIs))
}
//根据证书的URI获取信任域
trustDomain, err := GetTrustDomainFromURISAN(peerCert.URIs[0].String())
if err != nil {
return err
}
// 根据信任域获取对应的根证书
rootCertPool, ok := v.certPools[trustDomain]
if !ok {
return fmt.Errorf("no cert pool found for trust domain %s", trustDomain)
}
// 验证客户端证书
_, err = peerCert.Verify(x509.VerifyOptions{
Roots: rootCertPool,
Intermediates: intCertPool,
})
return err
}注册handler s.XDSServer.Register(s.secureGrpcServer)
StreamAggregatedResources 实现了envoy ADS接口
func (s *DiscoveryServer) StreamAggregatedResources(stream discovery.AggregatedDiscoveryService_StreamAggregatedResourcesServer) error {
// 检查服务器是否准备好接受客户端并处理新请求.当前准备就绪意味着缓存已同步,因此可以正确构建集群.如果不进行此检查,则下面的InitContext()调用将使用空配置初始化,从而导致重新连接的Envoy失去配置.这是除了添加caches之外的另一项安全检查。已将逻辑同步到就绪探针以处理kube-proxy ip表更新延迟的情况。
if !s.IsServerReady() {
return errors.New("server is not ready to serve discovery information")
}
ctx := stream.Context()
peerAddr := "0.0.0.0"
// 获取节点信息
if peerInfo, ok := peer.FromContext(ctx); ok {
peerAddr = peerInfo.Addr.String()
}
// 对节点身份进行验证
ids, err := s.authenticate(ctx)
if err != nil {
return err
}
if ids != nil {
adsLog.Debugf("Authenticated XDS: %v with identity %v", peerAddr, ids)
} else {
adsLog.Debuga("Unauthenticated XDS: ", peerAddr)
}
// 初始化上下文,因为启动了insecure,不确定哪一个先收到请求
if err = s.globalPushContext().InitContext(s.Env, nil, nil); err != nil {
adsLog.Warnf("Error reading config %v", err)
return err
}
con := newConnection(peerAddr, stream)
con.Identities = ids
// 从客户端接收信息
var receiveError error
reqChannel := make(chan *discovery.DiscoveryRequest, 1)
go s.receive(con, reqChannel, &receiveError)
for {
select {
// 读取请求
case req, ok := <-reqChannel:
if !ok {
return receiveError
}
// 返回信息给客户端
err := s.processRequest(req, con)
if err != nil {
return err
}
// 获取推送channnel的数据,eventhandler控制器将根据资源变更发送数据到pushChannel
case pushEv := <-con.pushChannel:
err := s.pushConnection(con, pushEv)
pushEv.done()
if err != nil {
return err
}
case <-con.stop:
return nil
}
}
}客户端身份认证
func (s *DiscoveryServer) authenticate(ctx context.Context) ([]string, error) {
if !features.XDSAuth {
return nil, nil
}
// 当前仅检查该请求是否具有使用我们的密钥签名的证书.受标志保护以避免破坏升级-应该在公开XDS的多集群/网格扩展中启用。
peerInfo, ok := peer.FromContext(ctx)
if !ok {
return nil, errors.New("invalid context")
}
// 不是TLS连接,我们将不执行身份验证,
if _, ok := peerInfo.AuthInfo.(credentials.TLSInfo); !ok {
return nil, nil
}
authFailMsgs := []string{}
// 遍历认证器,任何一个验证通过则认为通过
for _, authn := range s.Authenticators {
u, err := authn.Authenticate(ctx)
if u != nil && u.Identities != nil && err == nil {
return u.Identities, nil
}
....
}具体逻辑参见18节
处理请求的连接
func (s *DiscoveryServer) processRequest(req *discovery.DiscoveryRequest, con *Connection) error {
if s.StatusReporter != nil {
s.StatusReporter.RegisterEvent(con.ConID, req.TypeUrl, req.ResponseNonce)
}
if !s.shouldRespond(con, req) {
return nil
}
push := s.globalPushContext()
// 推送对应数据给客户端
return s.pushXds(con, push, versionInfo(), con.Watched(req.TypeUrl), &model.PushRequest{Full: true})
}主动推送
func (s *DiscoveryServer) pushConnection(con *Connection, pushEv *Event) error {
pushRequest := pushEv.pushRequest
// 全量推送更新节点当前的信息
if pushRequest.Full {
// Update Proxy with current information.
s.updateProxy(con.proxy, pushRequest.Push)
}
// 判断是否需要推送
if !ProxyNeedsPush(con.proxy, pushEv) {
adsLog.Debugf("Skipping push to %v, no updates required", con.ConID)
// 只有全量推送增加版本,增量推送不更新版本
if pushRequest.Full {
reportAllEvents(s.StatusReporter, con.ConID, pushRequest.Push.Version, nil)
}
return nil
}
currentVersion := versionInfo()
// 向所有生成器发送推送,每个生成器负责确定推送事件是否需要推送
for _, w := range getPushResources(con.proxy.WatchedResources) {
err := s.pushXds(con, pushRequest.Push, currentVersion, w, pushRequest)
if err != nil {
return err
}
}
if pushRequest.Full {
// 像unwatch资源报告所有事件.watch的资源将通过pushXds或ack报告。
reportAllEvents(s.StatusReporter, con.ConID, pushRequest.Push.Version, con.proxy.WatchedResources)
}
proxiesConvergeDelay.Record(time.Since(pushRequest.Start).Seconds())
return nil
}func (s *DiscoveryServer) pushXds(con *Connection, push *model.PushContext,
currentVersion string, w *model.WatchedResource, req *model.PushRequest) error {
if w == nil {
return nil
}
//根据请求url 确定Generator
gen := s.findGenerator(w.TypeUrl, con)
if gen == nil {
return nil
}
t0 := time.Now()
// 生成resource
cl := gen.Generate(con.proxy, push, w, req)
if cl == nil {
//如果没有内容推送,返回ACK
if s.StatusReporter != nil {
s.StatusReporter.RegisterEvent(con.ConID, w.TypeUrl, push.Version)
}
return nil
}
defer func() { recordPushTime(w.TypeUrl, time.Since(t0)) }()
resp := &discovery.DiscoveryResponse{
TypeUrl: w.TypeUrl,
VersionInfo: currentVersion,
Nonce: nonce(push.Version),
Resources: cl,
}
// 返回数据
err := con.send(resp)
if err != nil {
recordSendError(w.TypeUrl, con.ConID, err)
return err
}
if _, f := SkipLogTypes[w.TypeUrl]; !f {
adsLog.Infof("%s: PUSH for node:%s resources:%d", v3.GetShortType(w.TypeUrl), con.proxy.ID, len(cl))
}
return nil
}cds
func (c CdsGenerator) Generate(proxy *model.Proxy, push *model.PushContext, w *model.WatchedResource, req *model.PushRequest) model.Resources {
if !cdsNeedsPush(req, proxy) {
return nil
}
rawClusters := c.Server.ConfigGenerator.BuildClusters(proxy, push)
resources := model.Resources{}
for _, c := range rawClusters {
resources = append(resources, util.MessageToAny(c))
}
return resources
}eds
func (eds *EdsGenerator) Generate(proxy *model.Proxy, push *model.PushContext, w *model.WatchedResource, req *model.PushRequest) model.Resources {
if !edsNeedsPush(req.ConfigsUpdated) {
return nil
}
var edsUpdatedServices map[string]struct{}
if !req.Full {
edsUpdatedServices = model.ConfigNamesOfKind(req.ConfigsUpdated, gvk.ServiceEntry)
}
resources := make([]*any.Any, 0)
empty := 0
cached := 0
regenerated := 0
for _, clusterName := range w.ResourceNames {
if edsUpdatedServices != nil {
_, _, hostname, _ := model.ParseSubsetKey(clusterName)
if _, ok := edsUpdatedServices[string(hostname)]; !ok {
// Cluster was not updated, skip recomputing. This happens when we get an incremental update for a
// specific Hostname. On connect or for full push edsUpdatedServices will be empty.
continue
}
}
builder := NewEndpointBuilder(clusterName, proxy, push)
if marshalledEndpoint, f := eds.Server.Cache.Get(builder); f {
resources = append(resources, marshalledEndpoint)
cached++
} else {
l := eds.Server.generateEndpoints(builder)
if l == nil {
continue
}
regenerated++
if len(l.Endpoints) == 0 {
empty++
}
resource := util.MessageToAny(l)
resources = append(resources, resource)
eds.Server.Cache.Add(builder, resource)
}
}
if len(edsUpdatedServices) == 0 {
adsLog.Infof("EDS: PUSH for node:%s resources:%d empty:%v cached:%v/%v",
proxy.ID, len(resources), empty, cached, cached+regenerated)
} else {
adsLog.Debugf("EDS: PUSH INC for node:%s clusters:%d empty:%v cached:%v/%v",
proxy.ID, len(resources), empty, cached, cached+regenerated)
}
return resources
}lds
func (l LdsGenerator) Generate(proxy *model.Proxy, push *model.PushContext, w *model.WatchedResource, req *model.PushRequest) model.Resources {
if !ldsNeedsPush(req) {
return nil
}
listeners := l.Server.ConfigGenerator.BuildListeners(proxy, push)
resources := model.Resources{}
for _, c := range listeners {
resources = append(resources, util.MessageToAny(c))
}
return resources
}nds
func (n NdsGenerator) Generate(proxy *model.Proxy, push *model.PushContext, w *model.WatchedResource, req *model.PushRequest) model.Resources {
if !ndsNeedsPush(req) {
return nil
}
nt := n.Server.ConfigGenerator.BuildNameTable(proxy, push)
if nt == nil {
return nil
}
resources := model.Resources{util.MessageToAny(nt)}
return resources
}rds
func (c RdsGenerator) Generate(proxy *model.Proxy, push *model.PushContext, w *model.WatchedResource, req *model.PushRequest) model.Resources {
if !rdsNeedsPush(req) {
return nil
}
rawRoutes := c.Server.ConfigGenerator.BuildHTTPRoutes(proxy, push, w.ResourceNames)
resources := model.Resources{}
for _, c := range rawRoutes {
resources = append(resources, util.MessageToAny(c))
}
return resources
}Last updated