pkg/intercp/envoyadmin/forwarding_dds_client.go

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package envoyadmin

import (
	"context"
	"fmt"
	"reflect"
)

import (
	"github.com/pkg/errors"
)

import (
	mesh_proto "github.com/apache/dubbo-kubernetes/api/mesh/v1alpha1"
	"github.com/apache/dubbo-kubernetes/pkg/core"
	"github.com/apache/dubbo-kubernetes/pkg/core/admin"
	core_mesh "github.com/apache/dubbo-kubernetes/pkg/core/resources/apis/mesh"
	core_system "github.com/apache/dubbo-kubernetes/pkg/core/resources/apis/system"
	"github.com/apache/dubbo-kubernetes/pkg/core/resources/manager"
	core_model "github.com/apache/dubbo-kubernetes/pkg/core/resources/model"
	core_store "github.com/apache/dubbo-kubernetes/pkg/core/resources/store"
	"github.com/apache/dubbo-kubernetes/pkg/dds/service"
	"github.com/apache/dubbo-kubernetes/pkg/intercp/catalog"
)

var clientLog = core.Log.WithName("intercp").WithName("envoyadmin").WithName("client")

type NewClientFn = func(url string) (mesh_proto.InterCPEnvoyAdminForwardServiceClient, error)

type forwardingKdsEnvoyAdminClient struct {
	resManager     manager.ReadOnlyResourceManager
	cat            catalog.Catalog
	instanceID     string
	newClientFn    NewClientFn
	fallbackClient admin.EnvoyAdminClient
}

// NewForwardingEnvoyAdminClient returns EnvoyAdminClient which is only used on Global CP in multizone environment.
// It forwards the request to an instance of the Global CP to which Zone CP of given DPP is connected.
//
// For example:
// We have 2 instances of Global CP (ins-1, ins-2). Dataplane "backend" is in zone "east".
// The leader CP of zone "east" is connected to ins-1.
// If we execute config dump for "backend" on ins-1, we follow the regular flow of pkg/envoy/admin/kds_client.go
// If we execute config dump for "backend" on ins-2, we forward the request to ins-1 and then execute the regular flow.
func NewForwardingEnvoyAdminClient(
	resManager manager.ReadOnlyResourceManager,
	cat catalog.Catalog,
	instanceID string,
	newClientFn NewClientFn,
	fallbackClient admin.EnvoyAdminClient,
) admin.EnvoyAdminClient {
	return &forwardingKdsEnvoyAdminClient{
		resManager:     resManager,
		cat:            cat,
		instanceID:     instanceID,
		newClientFn:    newClientFn,
		fallbackClient: fallbackClient,
	}
}

var _ admin.EnvoyAdminClient = &forwardingKdsEnvoyAdminClient{}

func (f *forwardingKdsEnvoyAdminClient) PostQuit(context.Context, *core_mesh.DataplaneResource) error {
	panic("not implemented")
}

func (f *forwardingKdsEnvoyAdminClient) ConfigDump(ctx context.Context, proxy core_model.ResourceWithAddress) ([]byte, error) {
	instanceID, err := f.globalInstanceID(ctx, core_model.ZoneOfResource(proxy), service.ConfigDumpRPC)
	if err != nil {
		return nil, err
	}
	f.logIntendedAction(proxy, instanceID)
	if instanceID == f.instanceID {
		return f.fallbackClient.ConfigDump(ctx, proxy)
	}
	client, err := f.clientForInstanceID(ctx, instanceID)
	if err != nil {
		return nil, err
	}
	req := &mesh_proto.XDSConfigRequest{
		ResourceType: string(proxy.Descriptor().Name),
		ResourceName: proxy.GetMeta().GetName(),
		ResourceMesh: proxy.GetMeta().GetMesh(),
	}
	resp, err := client.XDSConfig(ctx, req)
	if err != nil {
		return nil, err
	}
	return resp.GetConfig(), nil
}

func (f *forwardingKdsEnvoyAdminClient) Stats(ctx context.Context, proxy core_model.ResourceWithAddress) ([]byte, error) {
	instanceID, err := f.globalInstanceID(ctx, core_model.ZoneOfResource(proxy), service.StatsRPC)
	if err != nil {
		return nil, err
	}
	f.logIntendedAction(proxy, instanceID)
	if instanceID == f.instanceID {
		return f.fallbackClient.Stats(ctx, proxy)
	}
	client, err := f.clientForInstanceID(ctx, instanceID)
	if err != nil {
		return nil, err
	}
	req := &mesh_proto.StatsRequest{
		ResourceType: string(proxy.Descriptor().Name),
		ResourceName: proxy.GetMeta().GetName(),
		ResourceMesh: proxy.GetMeta().GetMesh(),
	}
	resp, err := client.Stats(ctx, req)
	if err != nil {
		return nil, err
	}
	return resp.GetStats(), nil
}

func (f *forwardingKdsEnvoyAdminClient) Clusters(ctx context.Context, proxy core_model.ResourceWithAddress) ([]byte, error) {
	instanceID, err := f.globalInstanceID(ctx, core_model.ZoneOfResource(proxy), service.ClustersRPC)
	if err != nil {
		return nil, err
	}
	f.logIntendedAction(proxy, instanceID)
	if instanceID == f.instanceID {
		return f.fallbackClient.Clusters(ctx, proxy)
	}
	client, err := f.clientForInstanceID(ctx, instanceID)
	if err != nil {
		return nil, err
	}
	req := &mesh_proto.ClustersRequest{
		ResourceType: string(proxy.Descriptor().Name),
		ResourceName: proxy.GetMeta().GetName(),
		ResourceMesh: proxy.GetMeta().GetMesh(),
	}
	resp, err := client.Clusters(ctx, req)
	if err != nil {
		return nil, err
	}
	return resp.GetClusters(), nil
}

func (f *forwardingKdsEnvoyAdminClient) logIntendedAction(proxy core_model.ResourceWithAddress, instanceID string) {
	log := clientLog.WithValues(
		"name", proxy.GetMeta().GetName(),
		"mesh", proxy.GetMeta().GetMesh(),
		"type", proxy.Descriptor().Name,
		"instanceID", instanceID,
	)
	if instanceID == f.instanceID {
		log.V(1).Info("zone CP of the resource is connected to this Global CP instance. Executing operation")
	} else {
		log.V(1).Info("zone CP of the resource is connected to other Global CP instance. Forwarding the request")
	}
}

func (f *forwardingKdsEnvoyAdminClient) globalInstanceID(ctx context.Context, zone string, rpcName string) (string, error) {
	zoneInsightRes := core_system.NewZoneInsightResource()
	if err := f.resManager.Get(ctx, zoneInsightRes, core_store.GetByKey(zone, core_model.NoMesh)); err != nil {
		return "", err
	}
	streams := zoneInsightRes.Spec.GetEnvoyAdminStreams()
	var globalInstanceID string
	switch rpcName {
	case service.ConfigDumpRPC:
		globalInstanceID = streams.GetConfigDumpGlobalInstanceId()
	case service.StatsRPC:
		globalInstanceID = streams.GetStatsGlobalInstanceId()
	case service.ClustersRPC:
		globalInstanceID = streams.GetClustersGlobalInstanceId()
	default:
		return "", errors.Errorf("invalid operation %s", rpcName)
	}
	if globalInstanceID == "" {
		return "", &StreamNotConnectedError{rpcName: rpcName}
	}
	return globalInstanceID, nil
}

func (f *forwardingKdsEnvoyAdminClient) clientForInstanceID(ctx context.Context, instanceID string) (mesh_proto.InterCPEnvoyAdminForwardServiceClient, error) {
	instance, err := catalog.InstanceOfID(ctx, f.cat, instanceID)
	if err != nil {
		return nil, err
	}
	return f.newClientFn(instance.InterCpURL())
}

type StreamNotConnectedError struct {
	rpcName string
}

func (e *StreamNotConnectedError) Error() string {
	return fmt.Sprintf("stream to execute %s operations is not yet connected", e.rpcName)
}

func (e *StreamNotConnectedError) Is(err error) bool {
	return reflect.TypeOf(e) == reflect.TypeOf(err)
}