From d3ce94aa268439506f2a0c050fa751906759b26a Mon Sep 17 00:00:00 2001 From: Pascal Bleser Date: Thu, 2 Jul 2026 20:56:02 +0200 Subject: [PATCH] groupware: add body tracing to WS, fix metrics registrations, add operation identifiers for metrics --- pkg/jmap/api.go | 6 +- pkg/jmap/api_blob.go | 8 +- pkg/jmap/api_bootstrap.go | 2 +- pkg/jmap/api_calendar.go | 2 +- pkg/jmap/api_changes.go | 2 +- pkg/jmap/api_email.go | 28 +- pkg/jmap/api_identity.go | 2 +- pkg/jmap/api_mailbox.go | 6 +- pkg/jmap/api_objects.go | 2 +- pkg/jmap/api_vacation.go | 2 +- pkg/jmap/export_integration_test.go | 6 +- pkg/jmap/http.go | 279 +++++++++--------- pkg/jmap/model.go | 4 +- pkg/jmap/templates.go | 18 +- pkg/jmap/tools.go | 86 +++++- pkg/structs/structs.go | 8 + services/groupware/pkg/groupware/framework.go | 24 +- services/groupware/pkg/groupware/reva.go | 4 + services/groupware/pkg/metrics/metrics.go | 60 +++- 19 files changed, 335 insertions(+), 214 deletions(-) diff --git a/pkg/jmap/api.go b/pkg/jmap/api.go index b5910057e4..3d28996f05 100644 --- a/pkg/jmap/api.go +++ b/pkg/jmap/api.go @@ -26,7 +26,7 @@ func (c Context) WithContext(newContext context.Context) Context { } type ApiClient interface { - Command(request Request, ctx Context) (io.ReadCloser, Language, Error) + Command(operation Operation, request Request, ctx Context) (io.ReadCloser, Language, Error) io.Closer } @@ -50,8 +50,8 @@ type SessionClient interface { } type BlobClient interface { - UploadBinary(uploadUrl string, endpoint string, contentType string, content io.Reader, ctx Context) (UploadedBlob, Language, Error) - DownloadBinary(downloadUrl string, endpoint string, ctx Context) (*BlobDownload, Language, Error) + UploadBinary(uploadUrl string, operation Operation, endpoint string, contentType string, content io.Reader, ctx Context) (UploadedBlob, Language, Error) + DownloadBinary(downloadUrl string, operation Operation, endpoint string, ctx Context) (*BlobDownload, Language, Error) io.Closer } diff --git a/pkg/jmap/api_blob.go b/pkg/jmap/api_blob.go index 8d31d1e405..f772da217b 100644 --- a/pkg/jmap/api_blob.go +++ b/pkg/jmap/api_blob.go @@ -25,7 +25,7 @@ func (j *Client) GetBlobMetadata(accountId AccountId, ids []string, ctx Context) return ZeroResultV[BlobGetResponse](), jerr } - return command(j, ctx, cmd, func(body *Response) (BlobGetResponse, State, Error) { + return command(j, Operation("GetBlobMetadata"), ctx, cmd, func(body *Response) (BlobGetResponse, State, Error) { var response BlobGetResponse err := retrieveGet(ctx, body, get, "0", &response) if err != nil { @@ -48,7 +48,7 @@ func (j *Client) UploadBlobStream(accountId AccountId, contentType string, body uploadUrl := strings.NewReplacer( "{accountId}", url.PathEscape(string(accountId)), ).Replace(ctx.Session.UploadUrlTemplate) - return j.blob.UploadBinary(uploadUrl, ctx.Session.UploadEndpoint, contentType, body, ctx) + return j.blob.UploadBinary(uploadUrl, Operation("UploadBlobStream"), ctx.Session.UploadEndpoint, contentType, body, ctx) } func (j *Client) DownloadBlobStream(accountId AccountId, blobId string, name string, typ string, ctx Context) (*BlobDownload, Language, error) { //NOSONAR @@ -61,7 +61,7 @@ func (j *Client) DownloadBlobStream(accountId AccountId, blobId string, name str "{type}", url.PathEscape(typ), ).Replace(ctx.Session.DownloadUrlTemplate) logger = log.From(logger.With().Str(logDownloadUrl, downloadUrl).Str(logBlobId, blobId)) - return j.blob.DownloadBinary(downloadUrl, ctx.Session.DownloadEndpoint, ctx) + return j.blob.DownloadBinary(downloadUrl, Operation("DownloadBlobStream"), ctx.Session.DownloadEndpoint, ctx) } func (j *Client) UploadBlob(accountId AccountId, data []byte, contentType string, ctx Context) (Result[UploadedBlobWithHash], error) { @@ -97,7 +97,7 @@ func (j *Client) UploadBlob(accountId AccountId, data []byte, contentType string return ZeroResultV[UploadedBlobWithHash](), jerr } - return command(j, ctx, cmd, func(body *Response) (UploadedBlobWithHash, State, Error) { + return command(j, Operation("UploadBlob"), ctx, cmd, func(body *Response) (UploadedBlobWithHash, State, Error) { var uploadResponse BlobUploadResponse err := retrieveUpload(ctx, body, upload, "0", &uploadResponse) if err != nil { diff --git a/pkg/jmap/api_bootstrap.go b/pkg/jmap/api_bootstrap.go index b8d65be06c..3a0de312a4 100644 --- a/pkg/jmap/api_bootstrap.go +++ b/pkg/jmap/api_bootstrap.go @@ -27,7 +27,7 @@ func (j *Client) GetBootstrap(accountIds []AccountId, ctx Context) (Result[map[A if err != nil { return ZeroResultV[map[AccountId]AccountBootstrapResult](), err } - return command(j, ctx, cmd, func(body *Response) (map[AccountId]AccountBootstrapResult, State, Error) { + return command(j, Operation("GetBootstrap"), ctx, cmd, func(body *Response) (map[AccountId]AccountBootstrapResult, State, Error) { identityPerAccount := map[AccountId][]Identity{} quotaPerAccount := map[AccountId][]Quota{} identityStatesPerAccount := map[AccountId]State{} diff --git a/pkg/jmap/api_calendar.go b/pkg/jmap/api_calendar.go index 6516231a8b..5f52b0919c 100644 --- a/pkg/jmap/api_calendar.go +++ b/pkg/jmap/api_calendar.go @@ -13,7 +13,7 @@ func (j *Client) ParseICalendarBlob(accountId AccountId, blobIds []string, ctx C return ZeroResultV[CalendarEventParseResponse](), err } - return command(j, ctx, cmd, func(body *Response) (CalendarEventParseResponse, State, Error) { + return command(j, Operation("ParseICalendarBlob"), ctx, cmd, func(body *Response) (CalendarEventParseResponse, State, Error) { var response CalendarEventParseResponse err = retrieveParse(ctx, body, parse, "0", &response) if err != nil { diff --git a/pkg/jmap/api_changes.go b/pkg/jmap/api_changes.go index 7e868f0751..3ebed418ba 100644 --- a/pkg/jmap/api_changes.go +++ b/pkg/jmap/api_changes.go @@ -110,7 +110,7 @@ func (j *Client) GetChanges(accountId AccountId, stateMap StateMap, maxChanges u return ZeroResultV[ObjectChanges](), err } - return command(j, ctx, cmd, func(body *Response) (ObjectChanges, State, Error) { + return command(j, Operation("GetChanges"), ctx, cmd, func(body *Response) (ObjectChanges, State, Error) { changes := ObjectChanges{ MaxChanges: maxChanges, } diff --git a/pkg/jmap/api_email.go b/pkg/jmap/api_email.go index 2c588960bf..9860fdaedd 100644 --- a/pkg/jmap/api_email.go +++ b/pkg/jmap/api_email.go @@ -53,7 +53,7 @@ func (j *Client) GetEmails(accountId AccountId, ids []string, //NOSONAR if err != nil { return ZeroResultV[EmailGetResponse](), err } - return command(j, ctx, cmd, func(body *Response) (EmailGetResponse, State, Error) { + return command(j, Operation("GetEmails"), ctx, cmd, func(body *Response) (EmailGetResponse, State, Error) { if markAsSeen { var markResponse EmailSetResponse err = retrieveSet(ctx, body, markEmails, "0", &markResponse) @@ -91,7 +91,7 @@ func (j *Client) GetEmailBlobId(accountId AccountId, id string, ctx Context) (Re if err != nil { return ZeroResultV[string](), err } - return command(j, ctx, cmd, func(body *Response) (string, State, Error) { + return command(j, Operation("GetEmailBlobId"), ctx, cmd, func(body *Response) (string, State, Error) { var response EmailGetResponse err = retrieveGet(ctx, body, get, "0", &response) if err != nil { @@ -175,7 +175,7 @@ func (j *Client) GetAllEmailsInMailbox(accountId AccountId, mailboxId string, // return ZeroResultV[*EmailSearchResults](), err } - return command(j, ctx, cmd, func(body *Response) (*EmailSearchResults, State, Error) { + return command(j, Operation("GetAllEmailsInMailbox"), ctx, cmd, func(body *Response) (*EmailSearchResults, State, Error) { var queryResponse EmailQueryResponse err = retrieveQuery(ctx, body, query, "0", &queryResponse) if err != nil { @@ -262,7 +262,7 @@ func (j *Client) GetEmailChanges(accountId AccountId, return ZeroResultV[EmailChanges](), err } - return command(j, ctx, cmd, func(body *Response) (EmailChanges, State, Error) { + return command(j, Operation("GetEmailChanges"), ctx, cmd, func(body *Response) (EmailChanges, State, Error) { var changesResponse EmailChangesResponse err = retrieveChanges(ctx, body, changes, "0", &changesResponse) if err != nil { @@ -361,7 +361,7 @@ func (j *Client) QueryEmailSnippets(accountIds []AccountId, //NOSONAR return ZeroResultV[map[AccountId]EmailSnippetSearchResults](), err } - return command(j, ctx, cmd, func(body *Response) (map[AccountId]EmailSnippetSearchResults, State, Error) { + return command(j, Operation("QueryEmailSnippets"), ctx, cmd, func(body *Response) (map[AccountId]EmailSnippetSearchResults, State, Error) { results := make(map[AccountId]EmailSnippetSearchResults, len(uniqueAccountIds)) states := make(map[AccountId]State, len(uniqueAccountIds)) for _, accountId := range uniqueAccountIds { @@ -473,7 +473,7 @@ func (j *Client) QueryEmails(accountIds []AccountId, return ZeroResultV[map[AccountId]EmailSearchResults](), err } - return command(j, ctx, cmd, func(body *Response) (map[AccountId]EmailSearchResults, State, Error) { + return command(j, Operation("QueryEmails"), ctx, cmd, func(body *Response) (map[AccountId]EmailSearchResults, State, Error) { results := make(map[AccountId]EmailSearchResults, len(uniqueAccountIds)) queryStates := map[AccountId]State{} for _, accountId := range uniqueAccountIds { @@ -568,7 +568,7 @@ func (j *Client) QueryEmailsWithSnippets(accountIds []AccountId, //NOSONAR return ZeroResultV[map[AccountId]EmailQueryWithSnippetsResult](), err } - return command(j, ctx, cmd, func(body *Response) (map[AccountId]EmailQueryWithSnippetsResult, State, Error) { + return command(j, Operation("QueryEmailsWithSnippets"), ctx, cmd, func(body *Response) (map[AccountId]EmailQueryWithSnippetsResult, State, Error) { result := make(map[AccountId]EmailQueryWithSnippetsResult, len(uniqueAccountIds)) for _, accountId := range uniqueAccountIds { var queryResponse EmailQueryResponse @@ -662,7 +662,7 @@ func (j *Client) ImportEmail(accountId AccountId, data []byte, ctx Context) (Res return ZeroResultV[UploadedEmail](), err } - return command(j, ctx, cmd, func(body *Response) (UploadedEmail, State, Error) { + return command(j, Operation("ImportEmail"), ctx, cmd, func(body *Response) (UploadedEmail, State, Error) { var uploadResponse BlobUploadResponse err = retrieveResponseMatchParameters(ctx, body, CommandBlobUpload, "0", &uploadResponse) if err != nil { @@ -720,7 +720,7 @@ func (j *Client) CreateEmail(accountId AccountId, email EmailChange, replaceId s return ZeroResultV[*Email](), err } - return command(j, ctx, cmd, func(body *Response) (*Email, State, Error) { + return command(j, Operation("CreateEmail"), ctx, cmd, func(body *Response) (*Email, State, Error) { var setResponse EmailSetResponse err = retrieveResponseMatchParameters(ctx, body, CommandEmailSet, "0", &setResponse) if err != nil { @@ -767,7 +767,7 @@ func (j *Client) UpdateEmails(accountId AccountId, updates map[string]PatchObjec return ZeroResultV[map[string]*Email](), err } - return command(j, ctx, cmd, func(body *Response) (map[string]*Email, State, Error) { + return command(j, Operation("UpdateEmails"), ctx, cmd, func(body *Response) (map[string]*Email, State, Error) { var setResponse EmailSetResponse err = retrieveSet(ctx, body, set, "0", &setResponse) if err != nil { @@ -881,7 +881,7 @@ func (j *Client) SubmitEmail(accountId AccountId, identityId string, emailId str return ZeroResultV[EmailSubmission](), err } - return command(j, ctx, cmd, func(body *Response) (EmailSubmission, State, Error) { + return command(j, Operation("SubmitEmail"), ctx, cmd, func(body *Response) (EmailSubmission, State, Error) { var submissionResponse EmailSubmissionSetResponse err = retrieveSet(ctx, body, submit, "0", &submissionResponse) if err != nil { @@ -939,7 +939,7 @@ func (j *Client) GetEmailSubmissionStatus(accountId AccountId, submissionIds []s return ZeroResultV[EmailSubmissionGetResponse](), err } - return command(j, ctx, cmd, func(body *Response) (EmailSubmissionGetResponse, State, Error) { + return command(j, Operation("GetEmailSubmissionStatus"), ctx, cmd, func(body *Response) (EmailSubmissionGetResponse, State, Error) { var response EmailSubmissionGetResponse err = retrieveGet(ctx, body, get, "0", &response) if err != nil { @@ -981,7 +981,7 @@ func (j *Client) EmailsInThread(accountId AccountId, threadId string, return ZeroResultV[[]Email](), err } - return command(j, ctx, cmd, func(body *Response) ([]Email, State, Error) { + return command(j, Operation("EmailsInThread"), ctx, cmd, func(body *Response) ([]Email, State, Error) { var emailsResponse EmailGetResponse err = retrieveGet(ctx, body, get, "1", &emailsResponse) if err != nil { @@ -1070,7 +1070,7 @@ func (j *Client) QueryEmailSummaries(accountIds []AccountId, //NOSONAR return ZeroResultV[map[AccountId]EmailsSummary](), err } - return command(j, ctx, cmd, func(body *Response) (map[AccountId]EmailsSummary, State, Error) { + return command(j, Operation("QueryEmailSummaries"), ctx, cmd, func(body *Response) (map[AccountId]EmailsSummary, State, Error) { resp := map[AccountId]EmailsSummary{} for _, accountId := range uniqueAccountIds { var queryResponse EmailQueryResponse diff --git a/pkg/jmap/api_identity.go b/pkg/jmap/api_identity.go index a778373c93..13ef9cc17c 100644 --- a/pkg/jmap/api_identity.go +++ b/pkg/jmap/api_identity.go @@ -55,7 +55,7 @@ func (j *Client) GetIdentitiesAndMailboxes(mailboxAccountId AccountId, accountId if err != nil { return ZeroResultV[IdentitiesAndMailboxesGetResponse](), err } - return command(j, ctx, cmd, func(body *Response) (IdentitiesAndMailboxesGetResponse, State, Error) { + return command(j, Operation("GetIdentitiesAndMailboxes"), ctx, cmd, func(body *Response) (IdentitiesAndMailboxesGetResponse, State, Error) { identities := make(map[AccountId][]Identity, len(uniqueAccountIds)) stateByAccountId := make(map[AccountId]State, len(uniqueAccountIds)) notFound := []string{} diff --git a/pkg/jmap/api_mailbox.go b/pkg/jmap/api_mailbox.go index 2096da3f4e..9049c66445 100644 --- a/pkg/jmap/api_mailbox.go +++ b/pkg/jmap/api_mailbox.go @@ -55,7 +55,7 @@ func (j *Client) SearchMailboxes(accountIds []AccountId, filter MailboxFilterEle return ZeroResultV[map[AccountId][]Mailbox](), err } - return command(j, ctx, cmd, func(body *Response) (map[AccountId][]Mailbox, State, Error) { + return command(j, Operation("SearchMailboxes"), ctx, cmd, func(body *Response) (map[AccountId][]Mailbox, State, Error) { resp := map[AccountId][]Mailbox{} stateByAccountid := map[AccountId]State{} for _, accountId := range uniqueAccountIds { @@ -89,7 +89,7 @@ func (j *Client) SearchMailboxIdsPerRole(accountIds []AccountId, roles []string, return ZeroResultV[map[AccountId]map[string]string](), err } - return command(j, ctx, cmd, func(body *Response) (map[AccountId]map[string]string, State, Error) { + return command(j, Operation("SearchMailboxIdsPerRole"), ctx, cmd, func(body *Response) (map[AccountId]map[string]string, State, Error) { resp := map[AccountId]map[string]string{} stateByAccountid := map[AccountId]State{} for _, accountId := range uniqueAccountIds { @@ -228,7 +228,7 @@ func (j *Client) GetInboxNameForMultipleAccounts(accountIds []AccountId, ctx Con return ZeroResultV[map[AccountId]string](), err } - return command(j, ctx, cmd, func(body *Response) (map[AccountId]string, State, Error) { + return command(j, Operation("GetInboxNameForMultipleAccounts"), ctx, cmd, func(body *Response) (map[AccountId]string, State, Error) { resp := make(map[AccountId]string, n) stateByAccountId := make(map[AccountId]State, n) for _, accountId := range uniqueAccountIds { diff --git a/pkg/jmap/api_objects.go b/pkg/jmap/api_objects.go index d08b1beaac..5e863fa452 100644 --- a/pkg/jmap/api_objects.go +++ b/pkg/jmap/api_objects.go @@ -93,7 +93,7 @@ func (j *Client) GetObjects(accountId AccountId, //NOSONAR return ZeroResultV[Objects](), err } - return command(j, ctx, cmd, func(body *Response) (Objects, State, Error) { + return command(j, Operation("GetObjects"), ctx, cmd, func(body *Response) (Objects, State, Error) { objs := Objects{} states := map[string]State{} diff --git a/pkg/jmap/api_vacation.go b/pkg/jmap/api_vacation.go index c8146f83c6..792fa67704 100644 --- a/pkg/jmap/api_vacation.go +++ b/pkg/jmap/api_vacation.go @@ -98,7 +98,7 @@ func (j *Client) SetVacationResponse(accountId AccountId, vacation VacationRespo if err != nil { return ZeroResultV[VacationResponse](), err } - return command(j, ctx, cmd, func(body *Response) (VacationResponse, State, Error) { + return command(j, Operation("SetVacationResponse"), ctx, cmd, func(body *Response) (VacationResponse, State, Error) { var setResponse VacationResponseSetResponse err = retrieveSet(ctx, body, set, "0", &setResponse) if err != nil { diff --git a/pkg/jmap/export_integration_test.go b/pkg/jmap/export_integration_test.go index 83bed572b4..49cb392980 100644 --- a/pkg/jmap/export_integration_test.go +++ b/pkg/jmap/export_integration_test.go @@ -662,7 +662,7 @@ func createJmapClient(container *testcontainers.DockerContainer, ctx context.Con api := NewHttpJmapClient(&jh, auth, eventListener, true, 8192, true, 8192) - wscf, err := NewHttpWsClientFactory(wsd, auth, logger, eventListener) + wscf, err := NewHttpWsClientFactory(wsd, auth, logger, eventListener, true, 8192) if err != nil { return Client{}, nil, nil, err } @@ -676,6 +676,10 @@ type ContextPasswordAuthHttpJmapClientAuthenticator struct { var _ HttpJmapClientAuthenticator = &ContextPasswordAuthHttpJmapClientAuthenticator{} +func (h *ContextPasswordAuthHttpJmapClientAuthenticator) GetId() string { + return "context" +} + func (h *ContextPasswordAuthHttpJmapClientAuthenticator) Authenticate(ctx context.Context, username string, _ *oclog.Logger, req *http.Request) Error { password := ctx.Value(h.key).(string) req.SetBasicAuth(username, password) diff --git a/pkg/jmap/http.go b/pkg/jmap/http.go index aeecfd2d28..11937d9d3c 100644 --- a/pkg/jmap/http.go +++ b/pkg/jmap/http.go @@ -10,7 +10,6 @@ import ( "io" "net" "net/http" - "net/http/httputil" "net/url" "slices" "strconv" @@ -42,62 +41,64 @@ var ( ) const ( - logEndpoint = "endpoint" - logUri = "uri" - logMethod = "method" - logHttpStatus = "status" - logHttpStatusCode = "status-code" - logHttpUrl = "url" - logProto = "proto" - logProtoJmap = "jmap" - logProtoJmapWs = "jmapws" - logType = "type" - logTypeRequest = "request" - logTypeResponse = "response" - logTypePush = "push" - logDuration = "duration" - logBodyTruncated = "truncated" + logEndpoint = "endpoint" + logUri = "uri" + logMethod = "method" + logHttpStatus = "status" + logHttpStatusCode = "status-code" + logHttpUrl = "url" + logProto = "proto" + logProtoJmap = "jmap" + logProtoJmapWs = "jmapws" + logType = "type" + logTypeRequest = "request" + logTypeResponse = "response" + logTypePush = "push" + logDuration = "duration" + logBodyTruncated = "truncated" + logAuthenticatorId = "auth-id" ) // Record JMAP HTTP execution events that may occur, e.g. using metrics. type HttpJmapApiClientEventListener interface { - OnSuccessfulRequest(endpoint string, status int) - OnFailedRequest(endpoint string, err error) - OnFailedRequestWithStatus(endpoint string, status int) - OnResponseBodyReadingError(endpoint string, err error) - OnResponseBodyUnmarshallingError(endpoint string, err error) - OnSuccessfulWsRequest(endpoint string, status int) - OnFailedWsHandshakeRequestWithStatus(endpoint string, status int) + OnSuccessfulRequest(endpoint string, op Operation, status int) + OnFailedRequest(endpoint string, op Operation, err error) + OnFailedRequestWithStatus(endpoint string, op Operation, status int) + OnResponseBodyReadingError(endpoint string, op Operation, err error) + OnResponseBodyUnmarshallingError(endpoint string, op Operation, err error) + OnSuccessfulWsRequest(endpoint string, op Operation, status int) + OnFailedWsHandshakeRequestWithStatus(endpoint string, op Operation, status int) } type nullHttpJmapApiClientEventListener struct { } -func (l nullHttpJmapApiClientEventListener) OnSuccessfulRequest(endpoint string, status int) { +func (l nullHttpJmapApiClientEventListener) OnSuccessfulRequest(endpoint string, op Operation, status int) { // null implementation does nothing } -func (l nullHttpJmapApiClientEventListener) OnFailedRequest(endpoint string, err error) { +func (l nullHttpJmapApiClientEventListener) OnFailedRequest(endpoint string, op Operation, err error) { // null implementation does nothing } -func (l nullHttpJmapApiClientEventListener) OnFailedRequestWithStatus(endpoint string, status int) { +func (l nullHttpJmapApiClientEventListener) OnFailedRequestWithStatus(endpoint string, op Operation, status int) { // null implementation does nothing } -func (l nullHttpJmapApiClientEventListener) OnResponseBodyReadingError(endpoint string, err error) { +func (l nullHttpJmapApiClientEventListener) OnResponseBodyReadingError(endpoint string, op Operation, err error) { // null implementation does nothing } -func (l nullHttpJmapApiClientEventListener) OnResponseBodyUnmarshallingError(endpoint string, err error) { +func (l nullHttpJmapApiClientEventListener) OnResponseBodyUnmarshallingError(endpoint string, op Operation, err error) { // null implementation does nothing } -func (l nullHttpJmapApiClientEventListener) OnSuccessfulWsRequest(endpoint string, status int) { +func (l nullHttpJmapApiClientEventListener) OnSuccessfulWsRequest(endpoint string, op Operation, status int) { // null implementation does nothing } -func (l nullHttpJmapApiClientEventListener) OnFailedWsHandshakeRequestWithStatus(endpoint string, status int) { +func (l nullHttpJmapApiClientEventListener) OnFailedWsHandshakeRequestWithStatus(endpoint string, op Operation, status int) { // null implementation does nothing } var _ HttpJmapApiClientEventListener = nullHttpJmapApiClientEventListener{} type HttpJmapClientAuthenticator interface { + GetId() string Authenticate(ctx context.Context, username string, logger *log.Logger, req *http.Request) Error AuthenticateWS(ctx context.Context, username string, logger *log.Logger, headers http.Header) Error } @@ -130,6 +131,10 @@ func (h *MasterAuthHttpJmapClientAuthenticator) auth(username string, headers ht headers.Set("Authorization", sb.String()) } +func (h *MasterAuthHttpJmapClientAuthenticator) GetId() string { + return "master" +} + func (h *MasterAuthHttpJmapClientAuthenticator) Authenticate(_ context.Context, username string, _ *log.Logger, req *http.Request) Error { h.auth(username, req.Header) return nil @@ -177,8 +182,8 @@ func (e AuthenticationError) Unwrap() error { return e.Err } -func (h *HttpJmapClient) auth(ctx context.Context, username string, logger *log.Logger, req *http.Request) Error { - return h.authenticator.Authenticate(ctx, username, logger, req) +func (h *HttpJmapClient) auth(ctx context.Context, username string, logger *log.Logger, req *http.Request) (string, Error) { + return h.authenticator.GetId(), h.authenticator.Authenticate(ctx, username, logger, req) } var ( @@ -188,33 +193,16 @@ var ( func (h *HttpJmapClient) beforeRequest(_ context.Context, logger *log.Logger, _ string, endpoint string, req *http.Request) Error { l := logger.Trace() if h.traceRequests && l.Enabled() { - var logBuilder bytes.Buffer - for key, values := range req.Header { - if key == "Authorization" { - continue - } - for _, value := range values { - fmt.Fprintf(&logBuilder, "%s: %s\n", key, value) + if err := dumpHttpRequest(req, h.traceMaxRequestBodySize, func(method string, uri string, content string, truncated bool) { + if truncated { + l = l.Bool(logBodyTruncated, true) } + l.Str(logMethod, req.Method).Str(logUri, req.URL.String()). + Str(logEndpoint, endpoint).Str(logProto, logProtoJmap).Str(logType, logTypeRequest). + Msg(content) + }); err != nil { + return jmapError(err, JmapErrorRequestTracing) } - if h.traceMaxRequestBodySize >= 0 && req.Body != nil && req.Body != http.NoBody { - peekBuffer := new(bytes.Buffer) - limitedReader := io.LimitReader(req.Body, h.traceMaxRequestBodySize) - tee := io.TeeReader(limitedReader, peekBuffer) - if _, err := io.Copy(io.Discard, tee); err != nil { - return jmapError(fmt.Errorf("failed to peek at the request body for tracing: %v", err), JmapErrorRequestTracing) - } else { - logBuilder.Write(peekBuffer.Bytes()) - if int64(peekBuffer.Len()) >= h.traceMaxRequestBodySize { - l = l.Bool(logBodyTruncated, true) - } - fullBodyReader := io.MultiReader(peekBuffer, req.Body) - req.Body = io.NopCloser(fullBodyReader) - } - } - l.Str(logMethod, req.Method).Str(logUri, req.URL.String()). - Str(logEndpoint, endpoint).Str(logProto, logProtoJmap).Str(logType, logTypeRequest). - Msg(logBuilder.String()) } return nil } @@ -234,48 +222,15 @@ func (c httpResponseReadCloser) Close() error { func (h *HttpJmapClient) response(_ context.Context, logger *log.Logger, _ string, endpoint string, duration time.Duration, req *http.Request, resp *http.Response) (io.ReadCloser, error) { l := logger.Trace() if h.traceResponses && l.Enabled() { - var logBuilder bytes.Buffer - for key, values := range resp.Header { - for _, value := range values { - fmt.Fprintf(&logBuilder, "%s: %s\n", key, value) - } - } - - var peekBuffer *bytes.Buffer = nil - if h.traceMaxResponseBodySize > 0 { - peekBuffer = new(bytes.Buffer) - limitedReader := io.LimitReader(resp.Body, h.traceMaxResponseBodySize) - tee := io.TeeReader(limitedReader, peekBuffer) - if _, err := io.Copy(io.Discard, tee); err != nil { - if resp.Body != nil { - if err := resp.Body.Close(); err != nil { - logger.Error().Err(err).Msg("failed to close response body") //NOSONAR - } - } - return nil, err - } - - if int64(peekBuffer.Len()) >= h.traceMaxResponseBodySize { - l = l.Bool(logBodyTruncated, true) - } - - logBuilder.Write(peekBuffer.Bytes()) - } - - l.Str(logProto, logProtoJmap).Str(logType, logTypeResponse).Str(logEndpoint, endpoint). - Str(logMethod, req.Method).Str(logUri, req.URL.String()). - Str(logHttpStatus, log.SafeString(resp.Status)).Int(logHttpStatusCode, resp.StatusCode). - Dur(logDuration, duration). - Msg(logBuilder.String()) - - if peekBuffer != nil { - return httpResponseReadCloser{ - body: resp.Body, - Reader: io.MultiReader(peekBuffer, resp.Body), - }, nil - } else { - return resp.Body, nil - } + return dumpHttpResponse(req, resp, h.traceMaxResponseBodySize, func(method, uri, content string, truncated bool) { + l.Str(logProto, logProtoJmap).Str(logType, logTypeResponse).Str(logEndpoint, endpoint). + Str(logMethod, method).Str(logUri, uri). + Str(logHttpStatus, log.SafeString(resp.Status)).Int(logHttpStatusCode, resp.StatusCode). + Dur(logDuration, duration). + Msg(content) + }, func(err error) { + logger.Error().Err(err).Msg("failed to close response body") //NOSONAR + }) } else { return resp.Body, nil } @@ -309,15 +264,17 @@ func (h *HttpJmapClient) GetSession(ctx context.Context, sessionUrl *url.URL, us return SessionResponse{}, err } - if err := h.auth(ctx, username, logger, req); err != nil { + if authenticatorId, err := h.auth(ctx, username, logger, req); err != nil { return SessionResponse{}, err + } else { + logger = log.From(logger.With().Str(logAuthenticatorId, authenticatorId)) } before := time.Now() res, err := h.client.Do(req) duration := time.Since(before) if err != nil { - h.listener.OnFailedRequest(endpoint, err) + h.listener.OnFailedRequest(endpoint, Operation("GetSession"), err) logger.Error().Err(err).Msgf("failed to perform GET %v", sessionUrl) return SessionResponse{}, jmapError(err, JmapErrorInvalidHttpRequest) } @@ -334,29 +291,29 @@ func (h *HttpJmapClient) GetSession(ctx context.Context, sessionUrl *url.URL, us }() if res.StatusCode < 200 || res.StatusCode > 299 { - h.listener.OnFailedRequestWithStatus(endpoint, res.StatusCode) + h.listener.OnFailedRequestWithStatus(endpoint, Operation("GetSession"), res.StatusCode) logger.Error().Str(logHttpStatus, log.SafeString(res.Status)).Int(logHttpStatusCode, res.StatusCode).Msg("HTTP response status code is not 200") return SessionResponse{}, jmapError(fmt.Errorf("JMAP API response status is %v", res.Status), JmapErrorServerResponse) } - h.listener.OnSuccessfulRequest(endpoint, res.StatusCode) + h.listener.OnSuccessfulRequest(endpoint, Operation("GetSession"), res.StatusCode) if err != nil { logger.Error().Err(err).Msg("failed to read response body") //NOSONAR - h.listener.OnResponseBodyReadingError(endpoint, err) + h.listener.OnResponseBodyReadingError(endpoint, Operation("GetSession"), err) return SessionResponse{}, jmapError(err, JmapErrorReadingResponseBody) } var data SessionResponse if err := json.NewDecoder(body).Decode(&data); err != nil { logger.Error().Str(logHttpUrl, log.SafeString(sessionUrlStr)).Err(err).Msg("failed to decode JSON payload from .well-known/jmap response") - h.listener.OnResponseBodyUnmarshallingError(endpoint, err) + h.listener.OnResponseBodyUnmarshallingError(endpoint, Operation("GetSession"), err) return SessionResponse{}, jmapError(err, JmapErrorDecodingResponseBody) } return data, nil } -func (h *HttpJmapClient) Command(request Request, ctx Context) (io.ReadCloser, Language, Error) { //NOSONAR +func (h *HttpJmapClient) Command(operation Operation, request Request, ctx Context) (io.ReadCloser, Language, Error) { //NOSONAR session := ctx.Session logger := ctx.Logger acceptLanguage := ctx.AcceptLanguage @@ -389,15 +346,17 @@ func (h *HttpJmapClient) Command(request Request, ctx Context) (io.ReadCloser, L h.beforeRequest(cotx, logger, session.Username, endpoint, req) - if err := h.auth(cotx, session.Username, logger, req); err != nil { + if authenticatorId, err := h.auth(cotx, session.Username, logger, req); err != nil { return nil, "", err + } else { + logger = log.From(logger.With().Str(logAuthenticatorId, authenticatorId)) } before := time.Now() res, err := h.client.Do(req) duration := time.Since(before) if err != nil { - h.listener.OnFailedRequest(endpoint, err) + h.listener.OnFailedRequest(endpoint, operation, err) logger.Error().Err(err).Msgf("failed to perform POST %v", jmapUrl) return nil, "", jmapError(err, JmapErrorSendingRequest) } @@ -411,22 +370,22 @@ func (h *HttpJmapClient) Command(request Request, ctx Context) (io.ReadCloser, L language := Language(res.Header.Get("Content-Language")) //NOSONAR if err != nil { logger.Error().Err(err).Msg("failed to read response body") - h.listener.OnResponseBodyReadingError(endpoint, err) + h.listener.OnResponseBodyReadingError(endpoint, operation, err) return nil, language, jmapError(err, JmapErrorServerResponse) } if res.StatusCode < 200 || res.StatusCode > 299 { - h.listener.OnFailedRequestWithStatus(endpoint, res.StatusCode) + h.listener.OnFailedRequestWithStatus(endpoint, operation, res.StatusCode) logger.Error().Str(logEndpoint, endpoint).Str(logHttpStatus, log.SafeString(res.Status)).Msg("HTTP response status code is not 2xx") //NOSONAR return nil, language, jmapError(err, JmapErrorServerResponse) } - h.listener.OnSuccessfulRequest(endpoint, res.StatusCode) + h.listener.OnSuccessfulRequest(endpoint, operation, res.StatusCode) return body, language, nil } -func (h *HttpJmapClient) UploadBinary(uploadUrl string, endpoint string, contentType string, body io.Reader, ctx Context) (UploadedBlob, Language, Error) { //NOSONAR +func (h *HttpJmapClient) UploadBinary(uploadUrl string, operation Operation, endpoint string, contentType string, body io.Reader, ctx Context) (UploadedBlob, Language, Error) { //NOSONAR session := ctx.Session logger := ctx.Logger acceptLanguage := ctx.AcceptLanguage @@ -446,15 +405,17 @@ func (h *HttpJmapClient) UploadBinary(uploadUrl string, endpoint string, content } h.beforeRequest(cotx, logger, session.Username, endpoint, req) - if err := h.auth(cotx, session.Username, logger, req); err != nil { + if authenticatorId, err := h.auth(cotx, session.Username, logger, req); err != nil { return UploadedBlob{}, "", err + } else { + logger = log.From(logger.With().Str(logAuthenticatorId, authenticatorId)) } before := time.Now() res, err := h.client.Do(req) duration := time.Since(before) if err != nil { - h.listener.OnFailedRequest(endpoint, err) + h.listener.OnFailedRequest(endpoint, operation, err) logger.Error().Err(err).Msgf("failed to perform POST %v", uploadUrl) return UploadedBlob{}, "", jmapError(err, JmapErrorSendingRequest) } @@ -476,28 +437,28 @@ func (h *HttpJmapClient) UploadBinary(uploadUrl string, endpoint string, content language := Language(res.Header.Get("Content-Language")) if err != nil { logger.Error().Err(err).Msg("failed to read response body") - h.listener.OnResponseBodyReadingError(endpoint, err) + h.listener.OnResponseBodyReadingError(endpoint, operation, err) return UploadedBlob{}, language, jmapError(err, JmapErrorServerResponse) } if res.StatusCode < 200 || res.StatusCode > 299 { - h.listener.OnFailedRequestWithStatus(endpoint, res.StatusCode) + h.listener.OnFailedRequestWithStatus(endpoint, operation, res.StatusCode) logger.Error().Str(logHttpStatus, log.SafeString(res.Status)).Int(logHttpStatusCode, res.StatusCode).Msg("HTTP response status code is not 2xx") return UploadedBlob{}, language, jmapError(err, JmapErrorServerResponse) } - h.listener.OnSuccessfulRequest(endpoint, res.StatusCode) + h.listener.OnSuccessfulRequest(endpoint, operation, res.StatusCode) var result UploadedBlob if err := json.NewDecoder(responseBody).Decode(&result); err != nil { logger.Error().Str(logHttpUrl, log.SafeString(uploadUrl)).Err(err).Msg("failed to decode JSON payload from the upload response") - h.listener.OnResponseBodyUnmarshallingError(endpoint, err) + h.listener.OnResponseBodyUnmarshallingError(endpoint, operation, err) return UploadedBlob{}, language, jmapError(err, JmapErrorDecodingResponseBody) } return result, language, nil } -func (h *HttpJmapClient) DownloadBinary(downloadUrl string, endpoint string, ctx Context) (*BlobDownload, Language, Error) { //NOSONAR +func (h *HttpJmapClient) DownloadBinary(downloadUrl string, operation Operation, endpoint string, ctx Context) (*BlobDownload, Language, Error) { //NOSONAR session := ctx.Session logger := ctx.Logger acceptLanguage := ctx.AcceptLanguage @@ -516,15 +477,17 @@ func (h *HttpJmapClient) DownloadBinary(downloadUrl string, endpoint string, ctx } h.beforeRequest(cotx, logger, session.Username, endpoint, req) - if err := h.auth(cotx, session.Username, logger, req); err != nil { + if authenticatorId, err := h.auth(cotx, session.Username, logger, req); err != nil { return nil, "", err + } else { + logger = log.From(logger.With().Str(logAuthenticatorId, authenticatorId)) } before := time.Now() res, err := h.client.Do(req) duration := time.Since(before) if err != nil { - h.listener.OnFailedRequest(endpoint, err) + h.listener.OnFailedRequest(endpoint, operation, err) logger.Error().Err(err).Msgf("failed to perform GET %v", downloadUrl) return nil, "", jmapError(err, JmapErrorSendingRequest) } @@ -539,7 +502,7 @@ func (h *HttpJmapClient) DownloadBinary(downloadUrl string, endpoint string, ctx language := Language(res.Header.Get("Content-Language")) if err != nil { logger.Error().Err(err).Msg("failed to read response body") - h.listener.OnResponseBodyReadingError(endpoint, err) + h.listener.OnResponseBodyReadingError(endpoint, operation, err) return nil, language, jmapError(err, JmapErrorServerResponse) } @@ -547,11 +510,11 @@ func (h *HttpJmapClient) DownloadBinary(downloadUrl string, endpoint string, ctx return nil, language, nil } if res.StatusCode < 200 || res.StatusCode > 299 { - h.listener.OnFailedRequestWithStatus(endpoint, res.StatusCode) + h.listener.OnFailedRequestWithStatus(endpoint, operation, res.StatusCode) logger.Error().Str(logHttpStatus, log.SafeString(res.Status)).Int(logHttpStatusCode, res.StatusCode).Msg("HTTP response status code is not 2xx") return nil, language, jmapError(err, JmapErrorServerResponse) } - h.listener.OnSuccessfulRequest(endpoint, res.StatusCode) + h.listener.OnSuccessfulRequest(endpoint, operation, res.StatusCode) sizeStr := res.Header.Get("Content-Length") size := -1 @@ -604,16 +567,20 @@ type WebSocketPushDisable struct { } type HttpWsClientFactory struct { - dialer *websocket.Dialer - authenticator HttpJmapClientAuthenticator - logger *log.Logger - eventListener HttpJmapApiClientEventListener + dialer *websocket.Dialer + authenticator HttpJmapClientAuthenticator + logger *log.Logger + eventListener HttpJmapApiClientEventListener + traceWsResponses bool + traceMaxResponseBodySize int64 } var _ WsClientFactory = &HttpWsClientFactory{} func NewHttpWsClientFactory(dialer *websocket.Dialer, authenticator HttpJmapClientAuthenticator, logger *log.Logger, - eventListener HttpJmapApiClientEventListener) (*HttpWsClientFactory, error) { + eventListener HttpJmapApiClientEventListener, + traceWsResponses bool, traceMaxResponseBodySize int64, +) (*HttpWsClientFactory, error) { // RFC 8887: Section 4.2: // Otherwise, the client MUST make an authenticated HTTP request [RFC7235] on the encrypted connection // and MUST include the value "jmap" in the list of protocols for the "Sec-WebSocket-Protocol" header @@ -621,10 +588,12 @@ func NewHttpWsClientFactory(dialer *websocket.Dialer, authenticator HttpJmapClie dialer.Subprotocols = []string{"jmap"} return &HttpWsClientFactory{ - dialer: dialer, - authenticator: authenticator, - logger: logger, - eventListener: eventListener, + dialer: dialer, + authenticator: authenticator, + logger: logger, + eventListener: eventListener, + traceWsResponses: true, + traceMaxResponseBodySize: 4 * 1024, }, nil } @@ -632,9 +601,7 @@ func (w *HttpWsClientFactory) auth(ctx context.Context, username string, logger return w.authenticator.AuthenticateWS(ctx, username, logger, h) } -func (w *HttpWsClientFactory) connect(ctx context.Context, sessionProvider func() (*Session, error)) (*websocket.Conn, string, string, Error) { - logger := w.logger - +func (w *HttpWsClientFactory) connect(operation Operation, ctx context.Context, sessionProvider func() (*Session, error)) (*websocket.Conn, string, string, Error) { session, err := sessionProvider() if err != nil { return nil, "", "", jmapError(err, JmapErrorWssFailedToRetrieveSession) @@ -651,29 +618,49 @@ func (w *HttpWsClientFactory) connect(ctx context.Context, sessionProvider func( u := session.WebsocketUrl endpoint := session.WebsocketEndpoint + logger := log.From(w.logger.With().Str("username", log.SafeString(username)).Str("url", log.SafeString(u.String()))) + h := http.Header{} w.auth(ctx, username, logger, h) - w.logger.Trace().Str("username", log.SafeString(username)).Str("url", log.SafeString(u.String())).Msgf("connecting") + w.logger.Trace().Str("username", log.SafeString(username)).Str("url", log.SafeString(u.String())).Msgf("connecting") // TODO more/better attributes here for WS connection attempts + before := time.Now() c, res, err := w.dialer.DialContext(ctx, u.String(), h) + duration := time.Since(before) if err != nil { return nil, "", endpoint, jmapError(err, JmapErrorFailedToEstablishWssConnection) } - if w.logger.Trace().Enabled() { - responseBytes, err := httputil.DumpResponse(res, true) - if err == nil { - logger.Trace().Str(logEndpoint, endpoint).Str(logProto, logProtoJmapWs).Str(logType, logTypeResponse). - Str(logHttpStatus, log.SafeString(res.Status)).Int(logHttpStatusCode, res.StatusCode). - Msg(string(responseBytes)) + var body io.ReadCloser = nil + { + l := logger.Trace() + if w.traceWsResponses && l.Enabled() { + body, err = dumpHttpResponse(nil, res, w.traceMaxResponseBodySize, func(method, uri, content string, truncated bool) { + l.Str(logProto, logProtoJmap).Str(logType, logTypeResponse).Str(logEndpoint, endpoint). + Str(logMethod, method).Str(logUri, uri). + Str(logHttpStatus, log.SafeString(res.Status)).Int(logHttpStatusCode, res.StatusCode). + Dur(logDuration, duration). + Msg(content) + }, func(err error) { + logger.Error().Err(err).Msg("failed to close response body") //NOSONAR + }) } } + // we are not using the response body at all, at least for now + defer func() { + if body != nil { + if err := body.Close(); err != nil { + logger.Error().Err(err).Msg("failed to close response body") //NOSONAR + } + } + }() + if res.StatusCode != 101 { - w.eventListener.OnFailedRequestWithStatus(endpoint, res.StatusCode) + w.eventListener.OnFailedRequestWithStatus(endpoint, operation, res.StatusCode) logger.Error().Str(logHttpStatus, log.SafeString(res.Status)).Int(logHttpStatusCode, res.StatusCode).Msg("HTTP response status code is not 101") return nil, "", endpoint, jmapError(fmt.Errorf("JMAP WS API response status is %v", res.Status), JmapErrorServerResponse) } else { - w.eventListener.OnSuccessfulWsRequest(endpoint, res.StatusCode) + w.eventListener.OnSuccessfulWsRequest(endpoint, operation, res.StatusCode) } // RFC 8887: Section 4.2: @@ -748,7 +735,7 @@ func (w *HttpWsClient) readPump() { //NOSONAR } func (w *HttpWsClientFactory) EnableNotifications(ctx context.Context, pushState State, sessionProvider func() (*Session, error), listener WsPushListener) (WsClient, Error) { - c, username, endpoint, jerr := w.connect(ctx, sessionProvider) + c, username, endpoint, jerr := w.connect(Operation("EnableNotifications"), ctx, sessionProvider) if jerr != nil { return nil, jerr } diff --git a/pkg/jmap/model.go b/pkg/jmap/model.go index 3c617a802b..d814ac4b7f 100644 --- a/pkg/jmap/model.go +++ b/pkg/jmap/model.go @@ -1243,11 +1243,11 @@ type Response struct { // Clients may use this to detect if this object has changed and needs to be refetched. SessionState SessionState `json:"sessionState"` - // This MUST be the string "Response". + // Only for Websockets: This MUST be the string "Response". // The specification extends the Response object with two additional arguments when used over a WebSocket. Type TypeOfResponse `json:"@type,omitempty"` - // MUST be returned if an identifier is included in the request (optional). + // Only for Websockets: MUST be returned if an identifier is included in the request (optional). RequestId string `json:"requestId,omitempty"` } diff --git a/pkg/jmap/templates.go b/pkg/jmap/templates.go index 50e921a55b..49d2c513c7 100644 --- a/pkg/jmap/templates.go +++ b/pkg/jmap/templates.go @@ -22,7 +22,7 @@ func get[T Foo, GETREQ GetCommand[T], GETRESP GetResponse[T], ID any, RESP any]( return ZeroResultV[RESP](), err } - return command(client, ctx, cmd, func(body *Response) (RESP, State, Error) { + return command(client, Operation(name), ctx, cmd, func(body *Response) (RESP, State, Error) { var response GETRESP err = retrieveGet(ctx, body, get, "0", &response) if err != nil { @@ -73,7 +73,7 @@ func getN[T Foo, ITEM any, GETREQ GetCommand[T], GETRESP GetResponse[T], ID any, return ZeroResultV[RESP](), err } - return command(client, ctx, cmd, func(body *Response) (RESP, State, Error) { + return command(client, Operation(name), ctx, cmd, func(body *Response) (RESP, State, Error) { result := map[AccountId]ITEM{} responses := map[AccountId]GETRESP{} for _, accountId := range uniqueAccountIds { @@ -112,7 +112,7 @@ func create[T Foo, C any, SETREQ SetCommand[T], GETREQ GetCommand[T], SETRESP Se return ZeroResultV[*T](), err } - return command(client, ctx, cmd, func(body *Response) (*T, State, Error) { + return command(client, Operation(name), ctx, cmd, func(body *Response) (*T, State, Error) { var setResponse SETRESP err = retrieveSet(ctx, body, set, "0", &setResponse) if err != nil { @@ -165,7 +165,7 @@ func destroy[T Foo, REQ SetCommand[T], RESP SetResponse[T]](client *Client, name return ZeroResultV[map[string]SetError](), err } - return command(client, ctx, cmd, func(body *Response) (map[string]SetError, State, Error) { + return command(client, Operation(name), ctx, cmd, func(body *Response) (map[string]SetError, State, Error) { var setResponse RESP err = retrieveSet(ctx, body, set, "0", &setResponse) if err != nil { @@ -213,7 +213,7 @@ func changes[T Foo, CHANGESREQ ChangesCommand[T], GETREQ GetCommand[T], CHANGESR return ZeroResultV[RESP](), err } - return command(client, ctx, cmd, func(body *Response) (RESP, State, Error) { + return command(client, Operation(name), ctx, cmd, func(body *Response) (RESP, State, Error) { var changesResponse CHANGESRESP err = retrieveChanges(ctx, body, changes, "0", &changesResponse) if err != nil { @@ -302,7 +302,7 @@ func changesN[T Foo, CHANGESREQ ChangesCommand[T], GETREQ GetCommand[T], CHANGES return ZeroResultV[RESP](), err } - return command(client, ctx, cmd, func(body *Response) (RESP, State, Error) { + return command(client, Operation(name), ctx, cmd, func(body *Response) (RESP, State, Error) { changesItemByAccount := make(map[AccountId]CHANGESITEM, n) stateByAccountId := make(map[AccountId]State, n) for _, accountId := range uniqueAccountIds { @@ -357,7 +357,7 @@ func updates[T Foo, CHANGESREQ ChangesCommand[T], GETREQ GetCommand[T], CHANGESR return ZeroResultV[RESP](), err } - return command(client, ctx, cmd, func(body *Response) (RESP, State, Error) { + return command(client, Operation(name), ctx, cmd, func(body *Response) (RESP, State, Error) { var changesResponse CHANGESRESP err = retrieveChanges(ctx, body, changes, "0", &changesResponse) if err != nil { @@ -405,7 +405,7 @@ func update[T Foo, CHANGES Change[T], SET SetCommand[T], GET GetCommand[T], RESP return ZeroResultV[RESP](), err } - return command(client, ctx, cmd, func(body *Response) (RESP, State, Error) { + return command(client, Operation(name), ctx, cmd, func(body *Response) (RESP, State, Error) { var setResponse SETRESP err = retrieveSet(ctx, body, update, "0", &setResponse) if err != nil { @@ -529,7 +529,7 @@ func queryN[T Foo, FILTER any, SORT any, QUERY QueryCommand[T /*, QUERY*/], GET return ZeroResultV[map[AccountId]*RESP](), err } - return command(client, ctx, cmd, func(body *Response) (map[AccountId]*RESP, State, Error) { + return command(client, Operation(name), ctx, cmd, func(body *Response) (map[AccountId]*RESP, State, Error) { resp := map[AccountId]*RESP{} stateByAccountId := map[AccountId]State{} for accountId, queryParams := range accountIds { diff --git a/pkg/jmap/tools.go b/pkg/jmap/tools.go index ffa7ced3f8..0341433aff 100644 --- a/pkg/jmap/tools.go +++ b/pkg/jmap/tools.go @@ -1,9 +1,12 @@ package jmap import ( + "bytes" "encoding/json" "errors" "fmt" + "io" + "net/http" "reflect" "slices" "strings" @@ -62,7 +65,10 @@ type Cmdr interface { Hooks } +type Operation string + func command[T any](client Cmdr, //NOSONAR + operation Operation, ctx Context, request Request, mapper func(body *Response) (T, State, Error)) (Result[T], Error) { @@ -70,7 +76,7 @@ func command[T any](client Cmdr, //NOSONAR logger := ctx.Logger before := time.Now() - responseBody, language, jmapErr := client.Api().Command(request, ctx) + responseBody, language, jmapErr := client.Api().Command(operation, request, ctx) duration := time.Since(before) if jmapErr != nil { return ZeroResult[T](single(duration)), jmapErr @@ -445,3 +451,81 @@ func ns(namespaces ...JmapNamespace) []JmapNamespace { func ptr[T any | int | uint | bool | string](t T) *T { return &t } + +func dumpHttpRequest(req *http.Request, maxBodySize int64, closure func(method string, uri string, content string, truncated bool)) error { + var logBuilder bytes.Buffer + for key, values := range req.Header { + if key == "Authorization" { + continue + } + for _, value := range values { + fmt.Fprintf(&logBuilder, "%s: %s\n", key, value) + } + } + truncated := false + if maxBodySize >= 0 && req.Body != nil && req.Body != http.NoBody { + peekBuffer := new(bytes.Buffer) + limitedReader := io.LimitReader(req.Body, maxBodySize) + tee := io.TeeReader(limitedReader, peekBuffer) + if _, err := io.Copy(io.Discard, tee); err != nil { + return fmt.Errorf("failed to peek at the request body for tracing: %v", err) + } else { + logBuilder.Write(peekBuffer.Bytes()) + if int64(peekBuffer.Len()) >= maxBodySize { + truncated = true + } + fullBodyReader := io.MultiReader(peekBuffer, req.Body) + req.Body = io.NopCloser(fullBodyReader) + } + } + closure(req.Method, req.URL.String(), logBuilder.String(), truncated) + return nil +} + +func dumpHttpResponse(req *http.Request, resp *http.Response, maxBodySize int64, + closure func(method string, uri string, content string, truncated bool), + closeErrorClosure func(err error)) (io.ReadCloser, error) { + var logBuilder bytes.Buffer + for key, values := range resp.Header { + for _, value := range values { + fmt.Fprintf(&logBuilder, "%s: %s\n", key, value) + } + } + + var peekBuffer *bytes.Buffer = nil + truncated := false + if maxBodySize > 0 { + peekBuffer = new(bytes.Buffer) + limitedReader := io.LimitReader(resp.Body, maxBodySize) + tee := io.TeeReader(limitedReader, peekBuffer) + if _, err := io.Copy(io.Discard, tee); err != nil { + if resp.Body != nil { + if err := resp.Body.Close(); err != nil { + closeErrorClosure(err) + } + } + return nil, err + } + + if int64(peekBuffer.Len()) >= maxBodySize { + truncated = true + } + + logBuilder.Write(peekBuffer.Bytes()) + } + + if req != nil { + closure(req.Method, req.URL.String(), logBuilder.String(), truncated) + } else { + closure("", "", logBuilder.String(), truncated) + } + + if peekBuffer != nil { + return httpResponseReadCloser{ + body: resp.Body, + Reader: io.MultiReader(peekBuffer, resp.Body), + }, nil + } else { + return resp.Body, nil + } +} diff --git a/pkg/structs/structs.go b/pkg/structs/structs.go index 7f7d1c00e8..f2245f8bb8 100644 --- a/pkg/structs/structs.go +++ b/pkg/structs/structs.go @@ -170,6 +170,14 @@ func MapMap[A comparable, B any, X comparable, Y any](m map[A]B, mapper func(A, return r } +func FlatMap[K comparable, V any, E any](m map[K]V, mapper func(K, V) E) []E { + r := make([]E, len(m)) + for k, v := range m { + r = append(r, mapper(k, v)) + } + return r +} + // Creates a map from a map, keeping each key as-is, and using the mapper // function to determine the value to store into the resulting map. func MapValues[K comparable, S any, T any](m map[K]S, mapper func(S) T) map[K]T { diff --git a/services/groupware/pkg/groupware/framework.go b/services/groupware/pkg/groupware/framework.go index 45c5e750aa..099e583b0b 100644 --- a/services/groupware/pkg/groupware/framework.go +++ b/services/groupware/pkg/groupware/framework.go @@ -156,25 +156,26 @@ type groupwareHttpJmapApiClientMetricsRecorder struct { var _ jmap.HttpJmapApiClientEventListener = groupwareHttpJmapApiClientMetricsRecorder{} -func (r groupwareHttpJmapApiClientMetricsRecorder) OnSuccessfulRequest(endpoint string, status int) { +func (r groupwareHttpJmapApiClientMetricsRecorder) OnSuccessfulRequest(endpoint string, op jmap.Operation, status int) { r.m.SuccessfulRequestPerEndpointCounter.With(metrics.Endpoint(endpoint)).Inc() + r.m.OperationSpreadCounter.With(metrics.EndpointAndOperation(endpoint, op)).Inc() } -func (r groupwareHttpJmapApiClientMetricsRecorder) OnFailedRequest(endpoint string, err error) { +func (r groupwareHttpJmapApiClientMetricsRecorder) OnFailedRequest(endpoint string, op jmap.Operation, err error) { r.m.FailedRequestPerEndpointCounter.With(metrics.Endpoint(endpoint)).Inc() } -func (r groupwareHttpJmapApiClientMetricsRecorder) OnFailedRequestWithStatus(endpoint string, status int) { +func (r groupwareHttpJmapApiClientMetricsRecorder) OnFailedRequestWithStatus(endpoint string, op jmap.Operation, status int) { r.m.FailedRequestStatusPerEndpointCounter.With(metrics.EndpointAndStatus(endpoint, status)).Inc() } -func (r groupwareHttpJmapApiClientMetricsRecorder) OnResponseBodyReadingError(endpoint string, err error) { +func (r groupwareHttpJmapApiClientMetricsRecorder) OnResponseBodyReadingError(endpoint string, op jmap.Operation, err error) { r.m.ResponseBodyReadingErrorPerEndpointCounter.With(metrics.Endpoint(endpoint)).Inc() } -func (r groupwareHttpJmapApiClientMetricsRecorder) OnResponseBodyUnmarshallingError(endpoint string, err error) { +func (r groupwareHttpJmapApiClientMetricsRecorder) OnResponseBodyUnmarshallingError(endpoint string, op jmap.Operation, err error) { r.m.ResponseBodyUnmarshallingErrorPerEndpointCounter.With(metrics.Endpoint(endpoint)).Inc() } -func (r groupwareHttpJmapApiClientMetricsRecorder) OnSuccessfulWsRequest(endpoint string, status int) { +func (r groupwareHttpJmapApiClientMetricsRecorder) OnSuccessfulWsRequest(endpoint string, op jmap.Operation, status int) { // TODO metrics for WSS } -func (r groupwareHttpJmapApiClientMetricsRecorder) OnFailedWsHandshakeRequestWithStatus(endpoint string, status int) { +func (r groupwareHttpJmapApiClientMetricsRecorder) OnFailedWsHandshakeRequestWithStatus(endpoint string, op jmap.Operation, status int) { // TODO metrics for WSS } @@ -210,7 +211,10 @@ func NewGroupware(config *config.Config, logger *log.Logger, mux *chi.Mux, prome sanitize := true // TODO make configurable - m := metrics.New(prometheusRegistry, logger) + m, err := metrics.New(prometheusRegistry, logger) + if err != nil { + return nil, GroupwareInitializationError{Message: fmt.Sprintf("failed to configure metrics: %v", err), Err: err} + } userProvider := newRevaContextUsernameProvider() @@ -269,7 +273,9 @@ func NewGroupware(config *config.Config, logger *log.Logger, mux *chi.Mux, prome wsDialer.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} // #nosec G402 insecure TLS is a configuration option for development } - wsf, err = jmap.NewHttpWsClientFactory(wsDialer, auth, logger, jmapMetricsAdapter) + wsf, err = jmap.NewHttpWsClientFactory(wsDialer, auth, logger, jmapMetricsAdapter, + config.HTTP.TraceResponses, config.HTTP.TraceMaxResponseBodySize, + ) if err != nil { logger.Error().Err(err).Msg("failed to create websocket client") return nil, GroupwareInitializationError{Message: "failed to create websocket client", Err: err} diff --git a/services/groupware/pkg/groupware/reva.go b/services/groupware/pkg/groupware/reva.go index d7f3ef7f3c..bd8e224c8c 100644 --- a/services/groupware/pkg/groupware/reva.go +++ b/services/groupware/pkg/groupware/reva.go @@ -89,6 +89,10 @@ var tokenMissingInRevaContext = RevaError{ err: errors.New("token is missing from Reva context"), } +func (h *RevaBearerHttpJmapClientAuthenticator) GetId() string { + return "reva" +} + func (h *RevaBearerHttpJmapClientAuthenticator) Authenticate(ctx context.Context, _ string, logger *log.Logger, req *http.Request) jmap.Error { token, ok := revactx.ContextGetToken(ctx) if !ok { diff --git a/services/groupware/pkg/metrics/metrics.go b/services/groupware/pkg/metrics/metrics.go index 08828efdb7..af21a1c28d 100644 --- a/services/groupware/pkg/metrics/metrics.go +++ b/services/groupware/pkg/metrics/metrics.go @@ -1,10 +1,14 @@ package metrics import ( + "fmt" "reflect" "strconv" + "strings" + "github.com/opencloud-eu/opencloud/pkg/jmap" "github.com/opencloud-eu/opencloud/pkg/log" + "github.com/opencloud-eu/opencloud/pkg/structs" "github.com/prometheus/client_golang/prometheus" ) @@ -34,6 +38,7 @@ type Metrics struct { SSEEventsCounter *prometheus.CounterVec OutdatedSessionsCounter prometheus.Counter + OperationSpreadCounter *prometheus.CounterVec SuccessfulRequestPerEndpointCounter *prometheus.CounterVec FailedRequestPerEndpointCounter *prometheus.CounterVec FailedRequestStatusPerEndpointCounter *prometheus.CounterVec @@ -48,6 +53,7 @@ type Metrics struct { var Labels = struct { Endpoint string Result string + Operation string SessionCacheType string RequestId string TraceId string @@ -57,6 +63,7 @@ var Labels = struct { }{ Endpoint: "endpoint", Result: "result", + Operation: "op", SessionCacheType: "type", RequestId: "requestID", TraceId: "traceID", @@ -104,7 +111,7 @@ var Values = struct { } // New initializes the available metrics. -func New(registerer prometheus.Registerer, logger *log.Logger) *Metrics { +func New(registerer prometheus.Registerer, logger *log.Logger) (*Metrics, error) { m := &Metrics{ SessionCacheDesc: prometheus.NewDesc( prometheus.BuildFQName(Namespace, Subsystem, "session_cache"), @@ -178,11 +185,17 @@ func New(registerer prometheus.Registerer, logger *log.Logger) *Metrics { Name: "jmap_errors_count", Help: "Number of JMAP errors that occured", }, []string{Labels.Endpoint, Labels.ErrorCode}), + OperationSpreadCounter: prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: Namespace, + Subsystem: Subsystem, + Name: "jmap_operations", + Help: "Spread of operations by their names", + }, []string{Labels.Endpoint, Labels.Operation}), SuccessfulRequestPerEndpointCounter: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: Namespace, Subsystem: Subsystem, Name: "jmap_requests_count", - Help: "Number of JMAP requests", + Help: "Number of JMAP requests", //NOSONAR ConstLabels: prometheus.Labels{ Labels.Result: Values.Result.Success, }, @@ -191,7 +204,7 @@ func New(registerer prometheus.Registerer, logger *log.Logger) *Metrics { Namespace: Namespace, Subsystem: Subsystem, Name: "jmap_requests_count", - Help: "Number of JMAP requests", + Help: "Number of JMAP requests", //NOSONAR ConstLabels: prometheus.Labels{ Labels.Result: Values.Result.Failure, }, @@ -249,40 +262,51 @@ func New(registerer prometheus.Registerer, logger *log.Logger) *Metrics { }, []string{Labels.Endpoint}), } - registerAll(registerer, m, logger) + err := registerAll(registerer, m, logger) - return m + return m, err } func WithExemplar(obs prometheus.Observer, value float64, requestId string, traceId string) { obs.(prometheus.ExemplarObserver).ObserveWithExemplar(value, prometheus.Labels{Labels.RequestId: requestId, Labels.TraceId: traceId}) } -func registerAll(registerer prometheus.Registerer, m any, logger *log.Logger) { +func registerAll(registerer prometheus.Registerer, m any, logger *log.Logger) error { r := reflect.ValueOf(m) if r.Kind() == reflect.Pointer { r = r.Elem() } total := 0 - succeeded := 0 - failed := 0 + succeeded := []string{} + failed := map[string]error{} for i := 0; i < r.NumField(); i++ { n := r.Type().Field(i).Name f := r.Field(i) v := f.Interface() - c, ok := v.(prometheus.Collector) - if ok { + switch c := v.(type) { + case prometheus.Collector: total++ - err := registerer.Register(c) - if err != nil { - failed++ - logger.Warn().Err(err).Msgf("failed to register metric '%s' (%T)", n, c) + if err := registerer.Register(c); err != nil { + failed[n] = err } else { - succeeded++ + succeeded = append(succeeded, n) } + case *prometheus.Desc, prometheus.GaugeOpts, prometheus.CounterOpts, prometheus.UntypedOpts: + // skip these + default: + failed[n] = fmt.Errorf("unsupported metric '%s' of type %T", n, c) } } - logger.Debug().Msgf("registered %d/%d metrics successfully (%d failed)", succeeded, total, failed) + if len(failed) > 0 { + msg := strings.Join(structs.FlatMap(failed, func(name string, err error) string { + return fmt.Sprintf("'%s' (%v)", name, err) + }), ", ") + logger.Warn().Msgf("registered %d/%d metrics successfully (%d failed): %s", len(succeeded), total, len(failed), msg) + return fmt.Errorf("failed to register metrics: %s", msg) + } else { + logger.Debug().Msgf("registered %d/%d metrics successfully (%d failed)", len(succeeded), total, len(failed)) + return nil + } } type ConstMetricCollector struct { @@ -343,3 +367,7 @@ func Endpoint(endpoint string) prometheus.Labels { func EndpointAndStatus(endpoint string, status int) prometheus.Labels { return prometheus.Labels{Labels.Endpoint: endpoint, Labels.HttpStatusCode: strconv.Itoa(status)} } + +func EndpointAndOperation(endpoint string, operation jmap.Operation) prometheus.Labels { + return prometheus.Labels{Labels.Endpoint: endpoint, Labels.Operation: string(operation)} +}