diff --git a/go.mod b/go.mod index 4c6007f..ace7a58 100644 --- a/go.mod +++ b/go.mod @@ -41,7 +41,7 @@ require ( golang.org/x/crypto v0.14.0 golang.org/x/exp v0.0.0-20231006140011-7918f672742d golang.org/x/oauth2 v0.13.0 - google.golang.org/grpc v1.58.3 + google.golang.org/grpc v1.59.0 google.golang.org/protobuf v1.31.0 gopkg.in/yaml.v3 v3.0.1 gorm.io/driver/mysql v1.5.2 @@ -123,9 +123,9 @@ require ( golang.org/x/text v0.13.0 // indirect golang.org/x/tools v0.14.0 // indirect google.golang.org/appengine v1.6.8 // indirect - google.golang.org/genproto v0.0.0-20230711160842-782d3b101e98 // indirect - google.golang.org/genproto/googleapis/api v0.0.0-20230711160842-782d3b101e98 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20230711160842-782d3b101e98 // indirect + google.golang.org/genproto v0.0.0-20230822172742-b8732ec3820d // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20230822172742-b8732ec3820d // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d // indirect gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect modernc.org/libc v1.24.1 // indirect modernc.org/mathutil v1.6.0 // indirect diff --git a/go.sum b/go.sum index 52bf9f6..3fd9628 100644 --- a/go.sum +++ b/go.sum @@ -494,12 +494,20 @@ google.golang.org/appengine v1.6.8 h1:IhEN5q69dyKagZPYMSdIjS2HqprW324FRQZJcGqPAs google.golang.org/appengine v1.6.8/go.mod h1:1jJ3jBArFh5pcgW8gCtRJnepW8FzD1V44FJffLiz/Ds= google.golang.org/genproto v0.0.0-20230711160842-782d3b101e98 h1:Z0hjGZePRE0ZBWotvtrwxFNrNE9CUAGtplaDK5NNI/g= google.golang.org/genproto v0.0.0-20230711160842-782d3b101e98/go.mod h1:S7mY02OqCJTD0E1OiQy1F72PWFB4bZJ87cAtLPYgDR0= +google.golang.org/genproto v0.0.0-20230822172742-b8732ec3820d h1:VBu5YqKPv6XiJ199exd8Br+Aetz+o08F+PLMnwJQHAY= +google.golang.org/genproto v0.0.0-20230822172742-b8732ec3820d/go.mod h1:yZTlhN0tQnXo3h00fuXNCxJdLdIdnVFVBaRJ5LWBbw4= google.golang.org/genproto/googleapis/api v0.0.0-20230711160842-782d3b101e98 h1:FmF5cCW94Ij59cfpoLiwTgodWmm60eEV0CjlsVg2fuw= google.golang.org/genproto/googleapis/api v0.0.0-20230711160842-782d3b101e98/go.mod h1:rsr7RhLuwsDKL7RmgDDCUc6yaGr1iqceVb5Wv6f6YvQ= +google.golang.org/genproto/googleapis/api v0.0.0-20230822172742-b8732ec3820d h1:DoPTO70H+bcDXcd39vOqb2viZxgqeBeSGtZ55yZU4/Q= +google.golang.org/genproto/googleapis/api v0.0.0-20230822172742-b8732ec3820d/go.mod h1:KjSP20unUpOx5kyQUFa7k4OJg0qeJ7DEZflGDu2p6Bk= google.golang.org/genproto/googleapis/rpc v0.0.0-20230711160842-782d3b101e98 h1:bVf09lpb+OJbByTj913DRJioFFAjf/ZGxEz7MajTp2U= google.golang.org/genproto/googleapis/rpc v0.0.0-20230711160842-782d3b101e98/go.mod h1:TUfxEVdsvPg18p6AslUXFoLdpED4oBnGwyqk3dV1XzM= +google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d h1:uvYuEyMHKNt+lT4K3bN6fGswmK8qSvcreM3BwjDh+y4= +google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d/go.mod h1:+Bk1OCOj40wS2hwAMA+aCW9ypzm63QTBBHp6lQ3p+9M= google.golang.org/grpc v1.58.3 h1:BjnpXut1btbtgN/6sp+brB2Kbm2LjNXnidYujAVbSoQ= google.golang.org/grpc v1.58.3/go.mod h1:tgX3ZQDlNJGU96V6yHh1T/JeoBQ2TXdr43YbYSsCJk0= +google.golang.org/grpc v1.59.0 h1:Z5Iec2pjwb+LEOqzpB2MR12/eKFhDPhuqW91O+4bwUk= +google.golang.org/grpc v1.59.0/go.mod h1:aUPDwccQo6OTjy7Hct4AfBPD1GptF4fyUjIkQ9YtF98= google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= google.golang.org/protobuf v1.28.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= diff --git a/internal/conf/vendor.go b/internal/conf/vendor.go index f7a5649..47bff7e 100644 --- a/internal/conf/vendor.go +++ b/internal/conf/vendor.go @@ -21,10 +21,13 @@ type Etcd struct { } type Bilibili struct { - ServerName string `yaml:"server_name" env:"BILIBILI_SERVER_NAME"` - Endpoint string `yaml:"endpoint" env:"BILIBILI_ENDPOINT"` - JwtSecret string `yaml:"jwt_secret" env:"BILIBILI_JWT_SECRET"` - Scheme string `yaml:"scheme" lc:"grpc | http" env:"BILIBILI_SCHEME"` + ServerName string `yaml:"server_name" env:"BILIBILI_SERVER_NAME"` + Endpoint string `yaml:"endpoint" env:"BILIBILI_ENDPOINT"` + JwtSecret string `yaml:"jwt_secret" env:"BILIBILI_JWT_SECRET"` + Scheme string `yaml:"scheme" lc:"grpc | http" env:"BILIBILI_SCHEME"` + Tls bool `yaml:"tls" env:"BILIBILI_TLS"` + CustomCAFile string `yaml:"custom_ca_file" env:"BILIBILI_CUSTOM_CA_FILE"` + TimeOut string `yaml:"time_out" env:"BILIBILI_TIME_OUT"` Consul Consul `yaml:"consul,omitempty"` Etcd Etcd `yaml:"etcd,omitempty"` @@ -34,5 +37,6 @@ func DefaultBilibiliConfig() Bilibili { return Bilibili{ ServerName: "bilibili", Scheme: "grpc", + TimeOut: "5s", } } diff --git a/internal/vendor/bilibili.go b/internal/vendor/bilibili.go index 2f138d4..063c5c0 100644 --- a/internal/vendor/bilibili.go +++ b/internal/vendor/bilibili.go @@ -2,8 +2,11 @@ package vendor import ( "context" + "crypto/tls" + "crypto/x509" "errors" "fmt" + "os" "time" "github.com/go-kratos/aegis/circuitbreaker" @@ -12,8 +15,8 @@ import ( "github.com/go-kratos/kratos/contrib/registry/etcd/v2" "github.com/go-kratos/kratos/v2/middleware/auth/jwt" kcircuitbreaker "github.com/go-kratos/kratos/v2/middleware/circuitbreaker" - "github.com/go-kratos/kratos/v2/selector" - "github.com/go-kratos/kratos/v2/selector/wrr" + "google.golang.org/grpc" + ggrpc "github.com/go-kratos/kratos/v2/transport/grpc" "github.com/go-kratos/kratos/v2/transport/http" jwtv4 "github.com/golang-jwt/jwt/v4" @@ -31,7 +34,7 @@ func BilibiliClient() Bilibili { var ( bilibiliClient Bilibili - bilibiliDefaultClient = bilibiliService.NewBilibiliService(nil) + bilibiliDefaultClient Bilibili ) type Bilibili interface { @@ -63,147 +66,187 @@ var ( func InitBilibili(conf *conf.Bilibili) error { key := []byte(conf.JwtSecret) - selector.SetGlobalSelector(wrr.NewBuilder()) - if conf.Endpoint != "" { - switch conf.Scheme { - case "grpc": - con, err := ggrpc.DialInsecure( - context.Background(), - ggrpc.WithEndpoint(conf.Endpoint), - ggrpc.WithMiddleware( - jwt.Client(func(token *jwtv4.Token) (interface{}, error) { - return key, nil - }, jwt.WithSigningMethod(jwtv4.SigningMethodHS256)), - circuitBreaker, - ), - ) + bilibiliDefaultClient = bilibiliService.NewBilibiliService(nil) + switch conf.Scheme { + case "grpc": + opts := []ggrpc.ClientOption{} + + if conf.JwtSecret != "" { + opts = append(opts, ggrpc.WithMiddleware( + jwt.Client(func(token *jwtv4.Token) (interface{}, error) { + return key, nil + }, jwt.WithSigningMethod(jwtv4.SigningMethodHS256)), + circuitBreaker, + )) + } + + if conf.TimeOut != "" { + timeout, err := time.ParseDuration(conf.TimeOut) if err != nil { return err } - bilibiliClient = newGrpcBilibili(bilibili.NewBilibiliClient(con)) + opts = append(opts, ggrpc.WithTimeout(timeout)) + } + + if conf.Endpoint != "" { + opts = append(opts, ggrpc.WithEndpoint(conf.Endpoint)) log.Infof("bilibili client init success with endpoint: %s", conf.Endpoint) - case "http": - cli, err := http.NewClient( - context.Background(), - http.WithEndpoint(conf.Endpoint), - http.WithMiddleware( - jwt.Client(func(token *jwtv4.Token) (interface{}, error) { - return key, nil - }, jwt.WithSigningMethod(jwtv4.SigningMethodHS256)), - circuitBreaker, - ), - ) + } else if conf.Consul.Endpoint != "" { + if conf.ServerName == "" { + return errors.New("bilibili server name is empty") + } + c := api.DefaultConfig() + c.Address = conf.Consul.Endpoint + client, err := api.NewClient(c) if err != nil { return err } - bilibiliClient = newHTTPBilibili(bilibili.NewBilibiliHTTPClient(cli)) - default: - return errors.New("unknow bilibili scheme") - } - - } else if conf.Consul.Endpoint != "" { - if conf.ServerName == "" { - return errors.New("bilibili server name is empty") - } - c := api.DefaultConfig() - c.Address = conf.Consul.Endpoint - client, err := api.NewClient(c) - if err != nil { - panic(err) + endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName) + dis := consul.New(client) + opts = append(opts, ggrpc.WithEndpoint(endpoint), ggrpc.WithDiscovery(dis)) + log.Infof("bilibili client init success with consul: %s", conf.Consul.Endpoint) + } else if len(conf.Etcd.Endpoints) > 0 { + if conf.ServerName == "" { + return errors.New("bilibili server name is empty") + } + endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName) + cli, err := clientv3.New(clientv3.Config{ + Endpoints: conf.Etcd.Endpoints, + Username: conf.Etcd.Username, + Password: conf.Etcd.Password, + }) + if err != nil { + return err + } + dis := etcd.New(cli) + opts = append(opts, ggrpc.WithEndpoint(endpoint), ggrpc.WithDiscovery(dis)) + log.Infof("bilibili client init success with etcd: %v", conf.Etcd.Endpoints) + } else { + bilibiliClient = bilibiliDefaultClient + return nil } - endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName) - dis := consul.New(client) - switch conf.Scheme { - case "grpc": - con, err := ggrpc.DialInsecure( - context.Background(), - ggrpc.WithEndpoint(endpoint), - ggrpc.WithDiscovery(dis), - ggrpc.WithMiddleware( - jwt.Client(func(token *jwtv4.Token) (interface{}, error) { - return key, nil - }, jwt.WithSigningMethod(jwtv4.SigningMethodHS256)), - circuitBreaker, - ), - ) + var ( + con *grpc.ClientConn + err error + ) + if conf.Tls { + var rootCAs *x509.CertPool + rootCAs, err = x509.SystemCertPool() if err != nil { + fmt.Println("Failed to load system root CA:", err) return err } - bilibiliClient = newGrpcBilibili(bilibili.NewBilibiliClient(con)) - case "http": - cli, err := http.NewClient( + if conf.CustomCAFile != "" { + b, err := os.ReadFile(conf.CustomCAFile) + if err != nil { + panic(err) + } + rootCAs.AppendCertsFromPEM(b) + } + opts = append(opts, ggrpc.WithTLSConfig(&tls.Config{ + RootCAs: rootCAs, + })) + + con, err = ggrpc.Dial( + context.Background(), + opts..., + ) + } else { + con, err = ggrpc.DialInsecure( context.Background(), - http.WithEndpoint(endpoint), - http.WithDiscovery(dis), - http.WithMiddleware( - jwt.Client(func(token *jwtv4.Token) (interface{}, error) { - return key, nil - }, jwt.WithSigningMethod(jwtv4.SigningMethodHS256)), - circuitBreaker, - ), + opts..., ) + } + if err != nil { + return err + } + bilibiliClient = newGrpcBilibili(bilibili.NewBilibiliClient(con)) + case "http": + opts := []http.ClientOption{} + if conf.Tls { + rootCAs, err := x509.SystemCertPool() if err != nil { + fmt.Println("Failed to load system root CA:", err) return err } - bilibiliClient = newHTTPBilibili(bilibili.NewBilibiliHTTPClient(cli)) - default: - return errors.New("unknow bilibili scheme") + if conf.CustomCAFile != "" { + b, err := os.ReadFile(conf.CustomCAFile) + if err != nil { + panic(err) + } + rootCAs.AppendCertsFromPEM(b) + } + opts = append(opts, http.WithTLSConfig(&tls.Config{ + RootCAs: rootCAs, + })) } - log.Infof("bilibili client init success with consul: %s", conf.Consul.Endpoint) - } else if len(conf.Etcd.Endpoints) > 0 { - if conf.ServerName == "" { - return errors.New("bilibili server name is empty") + + if conf.JwtSecret != "" { + opts = append(opts, http.WithMiddleware( + jwt.Client(func(token *jwtv4.Token) (interface{}, error) { + return key, nil + }, jwt.WithSigningMethod(jwtv4.SigningMethodHS256)), + circuitBreaker, + )) } - endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName) - cli, err := clientv3.New(clientv3.Config{ - Endpoints: conf.Etcd.Endpoints, - Username: conf.Etcd.Username, - Password: conf.Etcd.Password, - }) - if err != nil { - return err + + if conf.TimeOut != "" { + timeout, err := time.ParseDuration(conf.TimeOut) + if err != nil { + return err + } + opts = append(opts, http.WithTimeout(timeout)) } - dis := etcd.New(cli) - switch conf.Scheme { - case "grpc": - con, err := ggrpc.DialInsecure( - context.Background(), - ggrpc.WithEndpoint(endpoint), - ggrpc.WithDiscovery(dis), - ggrpc.WithMiddleware( - jwt.Client(func(token *jwtv4.Token) (interface{}, error) { - return key, nil - }, jwt.WithSigningMethod(jwtv4.SigningMethodHS256)), - circuitBreaker, - ), - ) + + if conf.Endpoint != "" { + opts = append(opts, http.WithEndpoint(conf.Endpoint)) + log.Infof("bilibili client init success with endpoint: %s", conf.Endpoint) + } else if conf.Consul.Endpoint != "" { + if conf.ServerName == "" { + return errors.New("bilibili server name is empty") + } + c := api.DefaultConfig() + c.Address = conf.Consul.Endpoint + client, err := api.NewClient(c) if err != nil { return err } - bilibiliClient = newGrpcBilibili(bilibili.NewBilibiliClient(con)) - case "http": - cli, err := http.NewClient( - context.Background(), - http.WithEndpoint(endpoint), - http.WithDiscovery(dis), - http.WithMiddleware( - jwt.Client(func(token *jwtv4.Token) (interface{}, error) { - return key, nil - }, jwt.WithSigningMethod(jwtv4.SigningMethodHS256)), - circuitBreaker, - ), - ) + endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName) + dis := consul.New(client) + opts = append(opts, http.WithEndpoint(endpoint), http.WithDiscovery(dis)) + log.Infof("bilibili client init success with consul: %s", conf.Consul.Endpoint) + } else if len(conf.Etcd.Endpoints) > 0 { + if conf.ServerName == "" { + return errors.New("bilibili server name is empty") + } + endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName) + cli, err := clientv3.New(clientv3.Config{ + Endpoints: conf.Etcd.Endpoints, + Username: conf.Etcd.Username, + Password: conf.Etcd.Password, + }) if err != nil { return err } - bilibiliClient = newHTTPBilibili(bilibili.NewBilibiliHTTPClient(cli)) - default: - return errors.New("unknow bilibili scheme") + dis := etcd.New(cli) + opts = append(opts, http.WithEndpoint(endpoint), http.WithDiscovery(dis)) + log.Infof("bilibili client init success with etcd: %v", conf.Etcd.Endpoints) + } else { + bilibiliClient = bilibiliDefaultClient + return nil + } + con, err := http.NewClient( + context.Background(), + opts..., + ) + if err != nil { + return err } - log.Infof("bilibili client init success with etcd: %v", conf.Etcd.Endpoints) - } else { - bilibiliClient = bilibiliDefaultClient + bilibiliClient = newHTTPBilibili(bilibili.NewBilibiliHTTPClient(con)) + default: + return errors.New("unknow bilibili scheme") } + return nil } diff --git a/internal/vendor/vendor.go b/internal/vendor/vendor.go index 9303ad1..2460cbe 100644 --- a/internal/vendor/vendor.go +++ b/internal/vendor/vendor.go @@ -2,12 +2,15 @@ package vendor import ( klog "github.com/go-kratos/kratos/v2/log" + "github.com/go-kratos/kratos/v2/selector" + "github.com/go-kratos/kratos/v2/selector/wrr" log "github.com/sirupsen/logrus" "github.com/synctv-org/synctv/internal/conf" ) func Init(conf *conf.VendorConfig) error { klog.SetLogger(klog.NewStdLogger(log.StandardLogger().Writer())) + selector.SetGlobalSelector(wrr.NewBuilder()) if err := InitBilibili(&conf.Bilibili); err != nil { return err }