k8s-service/main.go

package main

import (
	"context"
	"fmt"
	grpcc "github.com/go-micro/plugins/v4/client/grpc"
	"github.com/go-micro/plugins/v4/registry/consul"
	grpcs "github.com/go-micro/plugins/v4/server/grpc"
	"github.com/go-micro/plugins/v4/wrapper/trace/opentelemetry"
	"github.com/google/uuid"
	"github.com/redis/go-redis/v9"
	"github.com/sirupsen/logrus"
	"go-micro.dev/v4"
	"go-micro.dev/v4/auth"
	"go-micro.dev/v4/registry"
	"go-micro.dev/v4/server"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/propagation"
	"gorm.io/driver/mysql"
	"gorm.io/gorm"
	"gorm.io/gorm/schema"
	req "sghgogs.com/micro/auth-service/domain/model/request"
	auth_service_pb "sghgogs.com/micro/auth-service/proto"
	"sghgogs.com/micro/k8s-service/config"
	"sghgogs.com/micro/k8s-service/handler/clusterrole"
	"sghgogs.com/micro/k8s-service/handler/clusterrolebinding"
	"sghgogs.com/micro/k8s-service/handler/configmap"
	"sghgogs.com/micro/k8s-service/handler/cronjob"
	"sghgogs.com/micro/k8s-service/handler/daemonset"
	"sghgogs.com/micro/k8s-service/handler/deployment"
	"sghgogs.com/micro/k8s-service/handler/ingress"
	"sghgogs.com/micro/k8s-service/handler/ingressroute"
	"sghgogs.com/micro/k8s-service/handler/job"
	"sghgogs.com/micro/k8s-service/handler/middleware"
	"sghgogs.com/micro/k8s-service/handler/namespace"
	"sghgogs.com/micro/k8s-service/handler/node"
	"sghgogs.com/micro/k8s-service/handler/persistentvolume"
	"sghgogs.com/micro/k8s-service/handler/persistentvolumeclaim"
	"sghgogs.com/micro/k8s-service/handler/pod"
	"sghgogs.com/micro/k8s-service/handler/role"
	"sghgogs.com/micro/k8s-service/handler/rolebinding"
	"sghgogs.com/micro/k8s-service/handler/secret"
	"sghgogs.com/micro/k8s-service/handler/service"
	"sghgogs.com/micro/k8s-service/handler/serviceaccount"
	"sghgogs.com/micro/k8s-service/handler/statefulset"
	pb "sghgogs.com/micro/k8s-service/proto"
	"sghgogs.com/micro/k8s-service/utils/authutil"
	utils_middleware "sghgogs.com/micro/k8s-service/utils/middleware"
	"sghgogs.com/micro/k8s-service/utils/tracing"
	"strings"
	"time"
)

var (
	name    = "kubernetesservice"
	version = "1.0.0"
)

func main() {
	if err := config.Load(); err != nil {
		logrus.Fatal(err)
	}
	// 1. 连接数据库
	var db *gorm.DB
	if cfg := config.DataBase(); cfg.Enable {
		address := fmt.Sprintf("%v:%v@(%v:%v)/%v?charset=utf8mb4,utf8&parseTime=True&loc=Local", cfg.Mysql.User, cfg.Mysql.Password, cfg.Mysql.Host, cfg.Mysql.Port, cfg.Mysql.DataBase)
		db, _ = gorm.Open(mysql.Open(address), &gorm.Config{
			// Logger: logger.Default.LogMode(logger.Info),
			NamingStrategy: schema.NamingStrategy{
				SingularTable: true,
			}})
	} else {
		// 没有配置数据库
		logrus.Info("There is no database configured")
	}

	if cfg := config.RedisAddress(); cfg.Enable {
		// UpdateRulesItems
		roles := make([]req.AdminRole, 0)
		db.Model(&req.AdminRole{}).Where("status = ?", auth_service_pb.StatusEnum_ENABLED).Preload("Permissions", "status = ?", auth_service_pb.StatusEnum_ENABLED).Find(&roles)
		authutil.NewJWTAuth(redis.NewClient(&redis.Options{
			Addr:     cfg.URL,      // Redis 服务器地址
			Password: cfg.Password, // Redis 密码，如果有的话
			DB:       0,            // 默认数据库
		}), name, cfg.Enable)
		authutil.JWTAuthService.SetRuleItems(UpdateRulesItems(roles))
		// cfg.Password
	}

	// 2. Create service
	srv := micro.NewService(
		micro.Server(grpcs.NewServer()),
		micro.Client(grpcc.NewClient()),
	)
	authService := authutil.JWTAuthService.Auth
	opts := []micro.Option{
		micro.Name(name),
		micro.Version(version),
		micro.Address(config.Address()),
		micro.Auth(
			authService,
		),
	}
	// 3.添加注册中心
	if cfg := config.Registry(); cfg.Enable {
		logrus.Info("添加注册中心")
		consul := consul.NewRegistry(func(options *registry.Options) {
			options.Addrs = []string{
				cfg.Consul.URL,
			}
		})
		opts = append(opts, micro.Registry(consul))
	}

	// 4.添加链路追踪
	if cfg := config.Tracing(); cfg.Enable {
		fmt.Println("加入进来了")
		tp, err := tracing.NewTracerProvider(name, version, srv.Server().Options().Id, cfg.Jaeger.URL)
		if err != nil {
			logrus.Fatal(err)
		}
		defer func() {
			ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
			defer cancel()
			if err = tp.Shutdown(ctx); err != nil {
				logrus.Fatal(err)
			}
		}()
		otel.SetTracerProvider(tp)
		otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{}))
		traceOpts := []opentelemetry.Option{
			opentelemetry.WithHandleFilter(func(ctx context.Context, r server.Request) bool {
				if e := r.Endpoint(); strings.HasPrefix(e, "Health.") {
					return true
				}
				return false
			}),
		}
		opts = append(opts, micro.WrapHandler(opentelemetry.NewHandlerWrapper(traceOpts...)))
	}
	opts = append(opts, micro.WrapHandler(utils_middleware.NewAuthWrapper(srv)))
	srv.Init(opts...)
	// 注册
	pb.RegisterClusterRoleServiceHandler(srv.Server(), &clusterrole.ClusterRole{})
	pb.RegisterClusterRoleBindingServiceHandler(srv.Server(), &clusterrolebinding.ClusterRoleBinding{})
	pb.RegisterConfigMapServiceHandler(srv.Server(), &configmap.ConfigMap{})
	pb.RegisterCronJobServiceHandler(srv.Server(), &cronjob.CronJob{})
	pb.RegisterDaemonSetServiceHandler(srv.Server(), &daemonset.DaemonSet{})
	pb.RegisterDeploymentServiceHandler(srv.Server(), &deployment.Deployment{})
	pb.RegisterIngressServiceHandler(srv.Server(), &ingress.Ingress{})
	pb.RegisterIngressRouteServiceHandler(srv.Server(), &ingressroute.IngressRoute{})
	pb.RegisterJobServiceHandler(srv.Server(), &job.Job{})
	pb.RegisterMiddlewareServiceHandler(srv.Server(), &middleware.Middleware{})
	pb.RegisterNamespaceServiceHandler(srv.Server(), &namespace.Namespace{})
	pb.RegisterNodeServiceHandler(srv.Server(), &node.Node{})
	pb.RegisterPersistentVolumeServiceHandler(srv.Server(), &persistentvolume.PersistentVolume{})
	pb.RegisterPersistentVolumeClaimServiceHandler(srv.Server(), &persistentvolumeclaim.PersistentVolumeClaim{})
	pb.RegisterPodServiceHandler(srv.Server(), &pod.Pod{})
	pb.RegisterRoleServiceHandler(srv.Server(), &role.Role{})
	pb.RegisterRoleBindingServiceHandler(srv.Server(), &rolebinding.RoleBinding{})
	pb.RegisterSecretServiceHandler(srv.Server(), &secret.Secret{})
	pb.RegisterServiceServiceHandler(srv.Server(), &service.Service{})
	pb.RegisterServiceAccountServiceHandler(srv.Server(), &serviceaccount.ServiceAccount{})
	pb.RegisterStatefulSetServiceHandler(srv.Server(), &statefulset.StatefulSet{})
	// Run service
	logrus.Info("Run service")
	if err := srv.Run(); err != nil {
		logrus.Fatal(err)
	}
}

func UpdateRulesItems(roles []req.AdminRole) []*auth.Rule {
	rules := make([]*auth.Rule, 0)
	for _, role := range roles {
		for _, permission := range role.Permissions {
			rules = append(rules, &auth.Rule{
				Resource: &auth.Resource{
					Name:     permission.Name,
					Type:     "user",
					Endpoint: permission.Endpoint,
				},
				ID:       uuid.New().String(),
				Scope:    role.Name,
				Priority: 1,
			})
		}
	}
	return rules
}