istio mcp-over-xds 原理及实现

// 初始化一个ads clientxdsMCP, err := adsc.New(srcAddress.Host, &adsc.Config{    Meta: model.NodeMetadata{        Generator: "api",    }.ToStruct(),    InitialDiscoveryRequests: adsc.ConfigInitialRequests(),})if err != nil {    return fmt.Errorf("failed to dial XDS %s %v", configSource.Address, err)}// 初始化一个configstorestore := memory.Make(collections.Pilot)// 初始化config controllerconfigController := memory.NewController(store)// 初始化istio config stroexdsMCP.Store = model.MakeIstioStore(configController)// 运行err = xdsMCP.Run()if err != nil {    return fmt.Errorf("MCP: failed running %v", err)}s.ConfigStores = append(s.ConfigStores, configController)log.Warn("Started XDS config ", s.ConfigStores)

out = append(out, &discovery.DiscoveryRequest{        // meshconfig类型        TypeUrl: collections.IstioMeshV1Alpha1MeshConfig.Resource().GroupVersionKind().String(),    })    for _, sch := range collections.Pilot.All() {        out = append(out, &discovery.DiscoveryRequest{            TypeUrl: sch.Resource().GroupVersionKind().String(),        })    }

    Pilot = collection.NewSchemasBuilder().        MustAdd(IstioNetworkingV1Alpha3Destinationrules).        MustAdd(IstioNetworkingV1Alpha3Envoyfilters).        MustAdd(IstioNetworkingV1Alpha3Gateways).        MustAdd(IstioNetworkingV1Alpha3Serviceentries).        MustAdd(IstioNetworkingV1Alpha3Sidecars).        MustAdd(IstioNetworkingV1Alpha3Virtualservices).        MustAdd(IstioNetworkingV1Alpha3Workloadentries).        MustAdd(IstioNetworkingV1Alpha3Workloadgroups).        MustAdd(IstioSecurityV1Beta1Authorizationpolicies).        MustAdd(IstioSecurityV1Beta1Peerauthentications).        MustAdd(IstioSecurityV1Beta1Requestauthentications).        Build()

func (a *ADSC) Run() error {    var err error    a.client = discovery.NewAggregatedDiscoveryServiceClient(a.conn)    a.stream, err = a.client.StreamAggregatedResources(context.Background())    if err != nil {        return err    }    a.sendNodeMeta = true    a.InitialLoad = 0    //  发送初始请求    for _, r := range a.cfg.InitialDiscoveryRequests {        if r.TypeUrl == v3.ClusterType {            a.watchTime = time.Now()        }        _ = a.Send(r)    }    a.RecvWg.Add(1)    // 接收ads server返回的数据    go a.handleRecv()    return nil}

groupVersionKind := config.GroupVersionKind{Group: gvk[0], Version: gvk[1], Kind: gvk[2]}

cfg := a.Store.Get(val.GroupVersionKind, val.Name, val.Namespace)        if cfg == nil {            _, err = a.Store.Create(*val)            if err != nil {                adscLog.Warnf("Error adding a new resource to the store %v", err)                continue            }        } else {            _, err = a.Store.Update(*val)            if err != nil {                adscLog.Warnf("Error updating an existing resource in the store %v", err)                continue            }        }

    configSources:      - address: xds://172.16.233.1:1109

kubectl apply -f demo.yaml

type AggregatedDiscoveryServiceServer interface {    // This is a gRPC-only API.    StreamAggregatedResources(AggregatedDiscoveryService_StreamAggregatedResourcesServer) error    DeltaAggregatedResources(AggregatedDiscoveryService_DeltaAggregatedResourcesServer) error}

func (m myserver) StreamAggregatedResources(stream d3.AggregatedDiscoveryService_StreamAggregatedResourcesServer) error {    if peerInfo, ok := peer.FromContext(stream.Context()); ok {        log.Println(peerInfo)    }    pushall(stream)    for {        select {        case <-m.psuhc:            pushall(stream)        }    }    return nil}

err = stream.Send(&d3.DiscoveryResponse{        TypeUrl:     "security.istio.io/v1beta1/PeerAuthentication",        VersionInfo: "1",        Nonce:       "",        Resources:   resp,    })

    pa := v1beta1.PeerAuthentication{        TypeMeta: v1.TypeMeta{            APIVersion: "security.istio.io/v1beta1",            Kind:       "PeerAuthentication",        },        ObjectMeta: v1.ObjectMeta{            Name:      "default",            Namespace: "istio-system",        },        Spec: securityv1beta1.PeerAuthentication{            Mtls: &securityv1beta1.PeerAuthentication_MutualTLS{                Mode: securityv1beta1.PeerAuthentication_MutualTLS_STRICT,            },        },    }

curl 172.17.116.27:8080/debug/configz[    {      "kind": "PeerAuthentication",      "apiVersion": "security.istio.io/v1beta1",      "metadata": {        "name": "default",        "namespace": "istio-system",        "resourceVersion": "2020-12-15 06:17:28.774277383 +0000 UTC m=+782.333181911",        "creationTimestamp": null      },      "spec": {        "mtls": {}      }    }  ]

kubectl create ns fookubectl apply -f <(istioctl kube-inject -f samples/httpbin/httpbin.yaml) -n fookubectl apply -f <(istioctl kube-inject -f samples/sleep/sleep.yaml) -n fookubectl create ns barkubectl apply -f <(istioctl kube-inject -f samples/httpbin/httpbin.yaml) -n barkubectl apply -f <(istioctl kube-inject -f samples/sleep/sleep.yaml) -n barkubectl create ns legacykubectl apply -f samples/httpbin/httpbin.yaml -n legacykubectl apply -f samples/sleep/sleep.yaml -n legacy

curl httpbin.foo:8000/ipcurl: (56) Recv failure: Connection reset by peer