TheArchive/zitadel
1
0
Fork 0
mirror of https://github.com/zitadel/zitadel.git synced 2025-08-12 00:47:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into new-eventstore

Browse Source
This commit is contained in:
adlerhurst
2020-11-23 11:40:50 +01:00
parent 4aadd290f4 855725c3c0
commit 6431fd2ec5
61 changed files with 449 additions and 982 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
internal/api/api.go
View File

@@ -16,6 +16,7 @@ import (
"github.com/caos/zitadel/internal/config/systemdefaults"
"github.com/caos/zitadel/internal/errors"
iam_model "github.com/caos/zitadel/internal/iam/model"
"github.com/caos/zitadel/internal/tracing"
)
type Config struct {
@@ -89,6 +90,7 @@ func (a *API) healthHandler() http.Handler {
handler := http.NewServeMux()
handler.HandleFunc("/healthz", handleHealth)
handler.HandleFunc("/ready", handleReadiness(checks))
handler.HandleFunc("/validate", handleValidate(checks))
handler.HandleFunc("/clientID", a.handleClientID)
return handler
@@ -96,17 +98,28 @@ func (a *API) healthHandler() http.Handler {
func handleHealth(w http.ResponseWriter, r *http.Request) {
_, err := w.Write([]byte("ok"))
logging.Log("API-Hfss2").OnError(err).Error("error writing ok for health")
logging.Log("API-Hfss2").OnError(err).WithField("traceID", tracing.TraceIDFromCtx(r.Context())).Error("error writing ok for health")
}
func handleReadiness(checks []ValidationFunction) func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
err := validate(r.Context(), checks)
if err == nil {
errors := validate(r.Context(), checks)
if len(errors) == 0 {
http_util.MarshalJSON(w, "ok", nil, http.StatusOK)
return
}
http_util.MarshalJSON(w, nil, err, http.StatusPreconditionFailed)
http_util.MarshalJSON(w, nil, errors[0], http.StatusPreconditionFailed)
}
}
func handleValidate(checks []ValidationFunction) func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
errors := validate(r.Context(), checks)
if len(errors) == 0 {
http_util.MarshalJSON(w, "ok", nil, http.StatusOK)
return
}
http_util.MarshalJSON(w, errors, nil, http.StatusOK)
}
}
@@ -121,12 +134,13 @@ func (a *API) handleClientID(w http.ResponseWriter, r *http.Request) {
type ValidationFunction func(ctx context.Context) error
func validate(ctx context.Context, validations []ValidationFunction) error {
func validate(ctx context.Context, validations []ValidationFunction) []error {
errors := make([]error, 0)
for _, validation := range validations {
if err := validation(ctx); err != nil {
logging.Log("API-vf823").WithError(err).Error("validation failed")
return err
logging.Log("API-vf823").WithError(err).WithField("traceID", tracing.TraceIDFromCtx(ctx)).Error("validation failed")
errors = append(errors, err)
}
}
return nil
return errors
}

11
internal/api/grpc/admin/probes.go
View File

@@ -4,19 +4,8 @@ import (
"context"
"github.com/golang/protobuf/ptypes/empty"
pb_struct "github.com/golang/protobuf/ptypes/struct"
"github.com/caos/zitadel/internal/errors"
)
func (s *Server) Healthz(_ context.Context, e *empty.Empty) (*empty.Empty, error) {
return &empty.Empty{}, nil
}
func (s *Server) Ready(ctx context.Context, e *empty.Empty) (*empty.Empty, error) {
return &empty.Empty{}, s.repo.Health(ctx)
}
func (s *Server) Validate(ctx context.Context, _ *empty.Empty) (*pb_struct.Struct, error) {
return nil, errors.ThrowUnimplemented(nil, "GRPC-98Gse", "Not implemented")
}

13
internal/api/grpc/auth/probes.go
View File

@@ -4,19 +4,8 @@ import (
"context"
"github.com/golang/protobuf/ptypes/empty"
pb_struct "github.com/golang/protobuf/ptypes/struct"
"github.com/caos/zitadel/internal/errors"
)
func (s *Server) Healthz(_ context.Context, e *empty.Empty) (*empty.Empty, error) {
return nil, errors.ThrowUnimplemented(nil, "GRPC-bst5W", "Not implemented")
}
func (s *Server) Ready(ctx context.Context, e *empty.Empty) (*empty.Empty, error) {
return nil, errors.ThrowUnimplemented(nil, "GRPC-or0vW", "Not implemented")
}
func (s *Server) Validate(ctx context.Context, _ *empty.Empty) (*pb_struct.Struct, error) {
return nil, errors.ThrowUnimplemented(nil, "GRPC-lo6Eg", "Not implemented")
return &empty.Empty{}, nil
}

4
internal/api/grpc/auth/user_converter.go
View File

@@ -3,6 +3,7 @@ package auth
import (
"context"
"encoding/json"
"github.com/caos/logging"
"github.com/golang/protobuf/ptypes"
"golang.org/x/text/language"
@@ -11,6 +12,7 @@ import (
"github.com/caos/zitadel/internal/api/authz"
"github.com/caos/zitadel/internal/eventstore/models"
"github.com/caos/zitadel/internal/tracing"
usr_model "github.com/caos/zitadel/internal/user/model"
"github.com/caos/zitadel/pkg/grpc/auth"
"github.com/caos/zitadel/pkg/grpc/message"
@@ -96,7 +98,7 @@ func profileViewFromModel(profile *usr_model.Profile) *auth.UserProfileView {
func updateProfileToModel(ctx context.Context, u *auth.UpdateUserProfileRequest) *usr_model.Profile {
preferredLanguage, err := language.Parse(u.PreferredLanguage)
logging.Log("GRPC-lk73L").OnError(err).Debug("language malformed")
logging.Log("GRPC-lk73L").OnError(err).WithField("traceID", tracing.TraceIDFromCtx(ctx)).Debug("language malformed")
return &usr_model.Profile{
ObjectRoot: models.ObjectRoot{AggregateID: authz.GetCtxData(ctx).UserID},

68
internal/api/grpc/client/middleware/tacing_test.go
View File

@@ -1,68 +0,0 @@
package middleware
import (
"context"
"testing"
"go.opencensus.io/plugin/ocgrpc"
"go.opencensus.io/trace"
"google.golang.org/grpc/stats"
)
func Test_tracingClientHandler_TagRPC(t *testing.T) {
type fields struct {
IgnoredMethods []GRPCMethod
ClientHandler ocgrpc.ClientHandler
}
type args struct {
ctx context.Context
tagInfo *stats.RPCTagInfo
}
tests := []struct {
name string
fields fields
args args
wantSpan bool
}{
{
"ignored method",
fields{
IgnoredMethods: []GRPCMethod{"ignore"},
ClientHandler: ocgrpc.ClientHandler{},
},
args{
ctx: context.Background(),
tagInfo: &stats.RPCTagInfo{
FullMethodName: "ignore",
},
},
false,
},
{
"tag",
fields{
IgnoredMethods: []GRPCMethod{"ignore"},
ClientHandler: ocgrpc.ClientHandler{},
},
args{
ctx: context.Background(),
tagInfo: &stats.RPCTagInfo{
FullMethodName: "tag",
},
},
true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
s := &tracingClientHandler{
IgnoredMethods: tt.fields.IgnoredMethods,
ClientHandler: tt.fields.ClientHandler,
}
got := s.TagRPC(tt.args.ctx, tt.args.tagInfo)
if (trace.FromContext(got) != nil) != tt.wantSpan {
t.Errorf("TagRPC() = %v, want %v", got, tt.wantSpan)
}
})
}
}

48
internal/api/grpc/client/middleware/tracing.go
View File

@@ -4,44 +4,32 @@ import (
"context"
"strings"
"go.opencensus.io/plugin/ocgrpc"
"go.opencensus.io/trace"
"google.golang.org/grpc"
"google.golang.org/grpc/stats"
grpc_utils "github.com/caos/zitadel/internal/api/grpc"
"github.com/caos/zitadel/internal/tracing"
grpc_trace "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
"google.golang.org/grpc"
)
type GRPCMethod string
func TracingStatsClient(ignoredMethods ...GRPCMethod) grpc.DialOption {
return grpc.WithStatsHandler(
&tracingClientHandler{
ignoredMethods,
ocgrpc.ClientHandler{
StartOptions: trace.StartOptions{
Sampler: tracing.Sampler(),
SpanKind: trace.SpanKindClient},
},
},
)
func DefaultTracingClient() grpc.UnaryClientInterceptor {
return TracingServer(grpc_utils.Healthz, grpc_utils.Readiness, grpc_utils.Validation)
}
func DefaultTracingStatsClient() grpc.DialOption {
return TracingStatsClient(grpc_utils.Healthz, grpc_utils.Readiness, grpc_utils.Validation)
}
func TracingServer(ignoredMethods ...GRPCMethod) grpc.UnaryClientInterceptor {
return func(
ctx context.Context,
method string,
req, reply interface{},
cc *grpc.ClientConn,
invoker grpc.UnaryInvoker,
opts ...grpc.CallOption,
) error {
type tracingClientHandler struct {
IgnoredMethods []GRPCMethod
ocgrpc.ClientHandler
}
func (s *tracingClientHandler) TagRPC(ctx context.Context, tagInfo *stats.RPCTagInfo) context.Context {
for _, method := range s.IgnoredMethods {
if strings.HasSuffix(tagInfo.FullMethodName, string(method)) {
return ctx
for _, ignoredMethod := range ignoredMethods {
if strings.HasSuffix(method, string(ignoredMethod)) {
return invoker(ctx, method, req, reply, cc, opts...)
}
}
return grpc_trace.UnaryClientInterceptor()(ctx, method, req, reply, cc, invoker, opts...)
}
return s.ClientHandler.TagRPC(ctx, tagInfo)
}

1
internal/api/grpc/errors/caos_errors.go
View File

@@ -2,6 +2,7 @@ package errors
import (
"context"
"github.com/caos/logging"
caos_errs "github.com/caos/zitadel/internal/errors"
"github.com/caos/zitadel/pkg/grpc/message"

12
internal/api/grpc/management/probes.go
View File

@@ -2,19 +2,9 @@ package management
import (
"context"
"github.com/caos/zitadel/internal/errors"
"github.com/golang/protobuf/ptypes/empty"
pb_struct "github.com/golang/protobuf/ptypes/struct"
)
func (s *Server) Healthz(_ context.Context, e *empty.Empty) (*empty.Empty, error) {
return nil, errors.ThrowUnimplemented(nil, "GRPC-mkd3y", "Not implemented")
}
func (s *Server) Ready(ctx context.Context, e *empty.Empty) (*empty.Empty, error) {
return nil, errors.ThrowUnimplemented(nil, "GRPC-pl6BM", "Not implemented")
}
func (s *Server) Validate(ctx context.Context, _ *empty.Empty) (*pb_struct.Struct, error) {
return nil, errors.ThrowUnimplemented(nil, "GRPC-2wxF", "Not implemented")
return &empty.Empty{}, nil
}

9
internal/api/grpc/server/gateway.go
View File

@@ -14,6 +14,7 @@ import (
client_middleware "github.com/caos/zitadel/internal/api/grpc/client/middleware"
http_util "github.com/caos/zitadel/internal/api/http"
http_mw "github.com/caos/zitadel/internal/api/http/middleware"
"github.com/caos/zitadel/internal/tracing"
)
const (
@@ -103,7 +104,7 @@ func createGateway(ctx context.Context, g Gateway, port string, customHeaders ..
mux := createMux(g, customHeaders...)
opts := createDialOptions(g)
err := g.RegisterGateway()(ctx, mux, http_util.Endpoint(port), opts)
logging.Log("SERVE-7B7G0E").OnError(err).Panic("failed to register grpc gateway")
logging.Log("SERVE-7B7G0E").OnError(err).WithField("traceID", tracing.TraceIDFromCtx(ctx)).Panic("failed to register grpc gateway")
return addInterceptors(mux, g)
}
@@ -116,8 +117,10 @@ func createMux(g Gateway, customHeaders ...string) *runtime.ServeMux {
}
func createDialOptions(g Gateway) []grpc.DialOption {
opts := []grpc.DialOption{grpc.WithInsecure()}
opts = append(opts, client_middleware.DefaultTracingStatsClient())
opts := []grpc.DialOption{
grpc.WithInsecure(),
grpc.WithUnaryInterceptor(client_middleware.DefaultTracingClient()),
}
if customOpts, ok := g.(gatewayCustomCallOptions); ok {
opts = append(opts, customOpts.GatewayCallOptions()...)

7
internal/api/grpc/server/middleware/auth_interceptor.go
View File

@@ -20,15 +20,14 @@ func AuthorizationInterceptor(verifier *authz.TokenVerifier, authConfig authz.Co
}
func authorize(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler, verifier *authz.TokenVerifier, authConfig authz.Config) (_ interface{}, err error) {
ctx, span := tracing.NewServerInterceptorSpan(ctx)
defer func() { span.EndWithError(err) }()
authOpt, needsToken := verifier.CheckAuthMethod(info.FullMethod)
if !needsToken {
span.End()
return handler(ctx, req)
}
ctx, span := tracing.NewServerInterceptorSpan(ctx)
defer func() { span.EndWithError(err) }()
authToken := grpc_util.GetAuthorizationHeader(ctx)
if authToken == "" {
return nil, status.Error(codes.Unauthenticated, "auth header missing")

47
internal/api/grpc/server/middleware/tracing.go
View File

@@ -4,45 +4,30 @@ import (
"context"
"strings"
"go.opencensus.io/plugin/ocgrpc"
"go.opencensus.io/trace"
"google.golang.org/grpc"
"google.golang.org/grpc/stats"
grpc_utils "github.com/caos/zitadel/internal/api/grpc"
"github.com/caos/zitadel/internal/tracing"
grpc_trace "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
"google.golang.org/grpc"
)
type GRPCMethod string
func TracingStatsServer(ignoredMethods ...GRPCMethod) grpc.ServerOption {
return grpc.StatsHandler(
&tracingServerHandler{
ignoredMethods,
ocgrpc.ServerHandler{
StartOptions: trace.StartOptions{
Sampler: tracing.Sampler(),
SpanKind: trace.SpanKindServer,
},
},
},
)
func DefaultTracingServer() grpc.UnaryServerInterceptor {
return TracingServer(grpc_utils.Healthz, grpc_utils.Readiness, grpc_utils.Validation)
}
func DefaultTracingStatsServer() grpc.ServerOption {
return TracingStatsServer(grpc_utils.Healthz, grpc_utils.Readiness, grpc_utils.Validation)
}
func TracingServer(ignoredMethods ...GRPCMethod) grpc.UnaryServerInterceptor {
return func(
ctx context.Context,
req interface{},
info *grpc.UnaryServerInfo,
handler grpc.UnaryHandler,
) (interface{}, error) {
type tracingServerHandler struct {
IgnoredMethods []GRPCMethod
ocgrpc.ServerHandler
}
func (s *tracingServerHandler) TagRPC(ctx context.Context, tagInfo *stats.RPCTagInfo) context.Context {
for _, method := range s.IgnoredMethods {
if strings.HasSuffix(tagInfo.FullMethodName, string(method)) {
return ctx
for _, ignoredMethod := range ignoredMethods {
if strings.HasSuffix(info.FullMethod, string(ignoredMethod)) {
return handler(ctx, req)
}
}
return grpc_trace.UnaryServerInterceptor()(ctx, req, info, handler)
}
return s.ServerHandler.TagRPC(ctx, tagInfo)
}

71
internal/api/grpc/server/middleware/tracing_test.go
View File

@@ -1,71 +0,0 @@
package middleware
import (
"context"
"testing"
"go.opencensus.io/plugin/ocgrpc"
"go.opencensus.io/trace"
"google.golang.org/grpc/stats"
)
func Test_tracingServerHandler_TagRPC(t *testing.T) {
type fields struct {
IgnoredMethods []GRPCMethod
ServerHandler ocgrpc.ServerHandler
}
type args struct {
ctx context.Context
tagInfo *stats.RPCTagInfo
}
type res struct {
wantSpan bool
}
tests := []struct {
name string
fields fields
args args
res res
}{
{
"ignored method",
fields{
IgnoredMethods: []GRPCMethod{"ignore"},
ServerHandler: ocgrpc.ServerHandler{},
},
args{
ctx: context.Background(),
tagInfo: &stats.RPCTagInfo{
FullMethodName: "ignore",
},
},
res{false},
},
{
"tag",
fields{
IgnoredMethods: []GRPCMethod{"ignore"},
ServerHandler: ocgrpc.ServerHandler{},
},
args{
ctx: context.Background(),
tagInfo: &stats.RPCTagInfo{
FullMethodName: "tag",
},
},
res{true},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
s := &tracingServerHandler{
IgnoredMethods: tt.fields.IgnoredMethods,
ServerHandler: tt.fields.ServerHandler,
}
got := s.TagRPC(tt.args.ctx, tt.args.tagInfo)
if (trace.FromContext(got) != nil) != tt.res.wantSpan {
t.Errorf("TagRPC() = %v, want %v", got, tt.res.wantSpan)
}
})
}
}

3
internal/api/grpc/server/probes.go
View File

@@ -9,6 +9,7 @@ import (
"github.com/caos/zitadel/internal/errors"
"github.com/caos/zitadel/internal/proto"
"github.com/caos/zitadel/internal/tracing"
)
type ValidationFunction func(ctx context.Context) error
@@ -41,7 +42,7 @@ func validate(ctx context.Context, validations map[string]ValidationFunction) ma
errors := make(map[string]error)
for id, validation := range validations {
if err := validation(ctx); err != nil {
logging.Log("API-vf823").WithError(err).Error("validation failed")
logging.Log("API-vf823").WithError(err).WithField("traceID", tracing.TraceIDFromCtx(ctx)).Error("validation failed")
errors[id] = err
}
}

7
internal/api/grpc/server/server.go
View File

@@ -11,6 +11,7 @@ import (
"github.com/caos/zitadel/internal/api/authz"
"github.com/caos/zitadel/internal/api/grpc/server/middleware"
"github.com/caos/zitadel/internal/api/http"
"github.com/caos/zitadel/internal/tracing"
)
const (
@@ -27,9 +28,9 @@ type Server interface {
func CreateServer(verifier *authz.TokenVerifier, authConfig authz.Config, lang language.Tag) *grpc.Server {
return grpc.NewServer(
middleware.DefaultTracingStatsServer(),
grpc.UnaryInterceptor(
grpc_middleware.ChainUnaryServer(
middleware.DefaultTracingServer(),
middleware.ErrorHandler(),
middleware.AuthorizationInterceptor(verifier, authConfig),
middleware.TranslationHandler(lang),
@@ -49,9 +50,9 @@ func Serve(ctx context.Context, server *grpc.Server, port string) {
go func() {
listener := http.CreateListener(port)
err := server.Serve(listener)
logging.Log("SERVE-Ga3e94").OnError(err).Panic("grpc server serve failed")
logging.Log("SERVE-Ga3e94").OnError(err).WithField("traceID", tracing.TraceIDFromCtx(ctx)).Panic("grpc server serve failed")
}()
logging.LogWithFields("SERVE-bZ44QM", "port", port).Info("grpc server is listening")
logging.LogWithFields("SERVE-bZ44QM", "port", port).WithField("traceID", tracing.TraceIDFromCtx(ctx)).Info("grpc server is listening")
}
func grpcPort(port string) string {

2
internal/api/http/middleware/trace_interceptor.go
View File

@@ -8,7 +8,7 @@ import (
)
func DefaultTraceHandler(handler http.Handler) http.Handler {
return tracing.TraceHandler(handler, http_utils.Probes...)
return TraceHandler(http_utils.Probes...)(handler)
}
func TraceHandler(ignoredMethods ...string) func(http.Handler) http.Handler {

7
internal/api/http/server.go
View File

@@ -5,6 +5,7 @@ import (
"net/http"
"github.com/caos/logging"
"github.com/caos/zitadel/internal/tracing"
)
func Serve(ctx context.Context, handler http.Handler, port, servername string) {
@@ -17,14 +18,14 @@ func Serve(ctx context.Context, handler http.Handler, port, servername string) {
go func() {
<-ctx.Done()
err := server.Shutdown(ctx)
logging.LogWithFields("HTTP-m7kBlq", "name", servername).OnError(err).Warnf("error during graceful shutdown of http server (%s)", servername)
logging.LogWithFields("HTTP-m7kBlq", "name", servername).WithField("traceID", tracing.TraceIDFromCtx(ctx)).OnError(err).Warnf("error during graceful shutdown of http server (%s)", servername)
}()
go func() {
err := server.Serve(listener)
logging.LogWithFields("HTTP-tBHR60", "name", servername).OnError(err).Panicf("http serve (%s) failed", servername)
logging.LogWithFields("HTTP-tBHR60", "name", servername).OnError(err).WithField("traceID", tracing.TraceIDFromCtx(ctx)).Panicf("http serve (%s) failed", servername)
}()
logging.LogWithFields("HTTP-KHh0Cb", "name", servername, "port", port).Infof("http server (%s) is listening", servername)
logging.LogWithFields("HTTP-KHh0Cb", "name", servername, "port", port).WithField("traceID", tracing.TraceIDFromCtx(ctx)).Infof("http server (%s) is listening", servername)
}
func RegisterHandler(mux *http.ServeMux, prefix string, handler http.Handler) {

5
internal/api/oidc/op.go
View File

@@ -12,6 +12,7 @@ import (
"github.com/caos/zitadel/internal/auth/repository"
"github.com/caos/zitadel/internal/config/types"
"github.com/caos/zitadel/internal/id"
"github.com/caos/zitadel/internal/tracing"
)
type OPHandlerConfig struct {
@@ -52,7 +53,7 @@ type OPStorage struct {
func NewProvider(ctx context.Context, config OPHandlerConfig, repo repository.Repository, localDevMode bool) op.OpenIDProvider {
cookieHandler, err := middleware.NewUserAgentHandler(config.UserAgentCookieConfig, id.SonyFlakeGenerator, localDevMode)
logging.Log("OIDC-sd4fd").OnError(err).Panic("cannot user agent handler")
logging.Log("OIDC-sd4fd").OnError(err).WithField("traceID", tracing.TraceIDFromCtx(ctx)).Panic("cannot user agent handler")
config.OPConfig.CodeMethodS256 = true
provider, err := op.NewOpenIDProvider(
ctx,
@@ -71,7 +72,7 @@ func NewProvider(ctx context.Context, config OPHandlerConfig, repo repository.Re
op.WithCustomKeysEndpoint(op.NewEndpointWithURL(config.Endpoints.Keys.Path, config.Endpoints.Keys.URL)),
op.WithRetry(3, time.Duration(30*time.Second)),
)
logging.Log("OIDC-asf13").OnError(err).Panic("cannot create provider")
logging.Log("OIDC-asf13").OnError(err).WithField("traceID", tracing.TraceIDFromCtx(ctx)).Panic("cannot create provider")
return provider
}