From 477b541bf0855821ac2f1d9bf693534978e01696 Mon Sep 17 00:00:00 2001 From: Jim Date: Thu, 5 Aug 2021 10:54:10 -0400 Subject: [PATCH] Events: ongoing clean up of eventing (#1448) --- internal/db/read_writer.go | 96 +++++++++---------- internal/errors/error.go | 17 ++-- internal/scheduler/job/repository_job.go | 10 +- .../handlers/workers/worker_service.go | 27 ------ internal/servers/controller/tickers.go | 1 - internal/servers/worker/handler.go | 12 --- internal/servers/worker/session.go | 8 +- internal/servers/worker/status.go | 1 - internal/servers/worker/tcp_proxy.go | 6 +- 9 files changed, 65 insertions(+), 113 deletions(-) diff --git a/internal/db/read_writer.go b/internal/db/read_writer.go index e13373ba3..087962e43 100644 --- a/internal/db/read_writer.go +++ b/internal/db/read_writer.go @@ -213,7 +213,7 @@ func (rw *Db) Exec(ctx context.Context, sql string, values []interface{}, _ ...O } gormDb := rw.underlying.Exec(sql, values...) if gormDb.Error != nil { - return NoRowsAffected, errors.Wrap(ctx, gormDb.Error, op) + return NoRowsAffected, errors.Wrap(ctx, gormDb.Error, op, errors.WithoutEvent()) } return int(gormDb.RowsAffected), nil } @@ -229,7 +229,7 @@ func (rw *Db) Query(ctx context.Context, sql string, values []interface{}, _ ... } gormDb := rw.underlying.Raw(sql, values...) if gormDb.Error != nil { - return nil, errors.Wrap(ctx, gormDb.Error, op) + return nil, errors.Wrap(ctx, gormDb.Error, op, errors.WithoutEvent()) } return gormDb.Rows() } @@ -255,7 +255,7 @@ func (rw *Db) lookupAfterWrite(ctx context.Context, i interface{}, opt ...Option return nil } if err := rw.LookupById(ctx, i, opt...); err != nil { - return errors.Wrap(ctx, err, op) + return errors.Wrap(ctx, err, op, errors.WithoutEvent()) } return nil } @@ -304,7 +304,7 @@ func (rw *Db) Create(ctx context.Context, i interface{}, opt ...Option) error { } } if err := rw.underlying.Create(i).Error; err != nil { - return errors.Wrap(ctx, err, op, errors.WithMsg("create failed")) + return errors.Wrap(ctx, err, op, errors.WithMsg("create failed"), errors.WithoutEvent()) } if withOplog { if err := rw.addOplog(ctx, CreateOp, opts, ticket, i); err != nil { @@ -319,7 +319,7 @@ func (rw *Db) Create(ctx context.Context, i interface{}, opt ...Option) error { *opts.newOplogMsg = *msg } if err := rw.lookupAfterWrite(ctx, i, opt...); err != nil { - return errors.Wrap(ctx, err, op) + return errors.Wrap(ctx, err, op, errors.WithoutEvent()) } return nil } @@ -369,7 +369,7 @@ func (rw *Db) CreateItems(ctx context.Context, createItems []interface{}, opt .. } for _, item := range createItems { if err := rw.Create(ctx, item); err != nil { - return errors.Wrap(ctx, err, op) + return errors.Wrap(ctx, err, op, errors.WithoutEvent()) } } if opts.withOplog { @@ -496,9 +496,9 @@ func (rw *Db) Update(ctx context.Context, i interface{}, fieldMaskPaths []string } if underlying.Error != nil { if underlying.Error == gorm.ErrRecordNotFound { - return NoRowsAffected, errors.E(ctx, errors.WithCode(errors.RecordNotFound), errors.WithOp(op)) + return NoRowsAffected, errors.E(ctx, errors.WithCode(errors.RecordNotFound), errors.WithOp(op), errors.WithoutEvent()) } - return NoRowsAffected, errors.Wrap(ctx, underlying.Error, op) + return NoRowsAffected, errors.Wrap(ctx, underlying.Error, op, errors.WithoutEvent()) } rowsUpdated := int(underlying.RowsAffected) if rowsUpdated > 0 && (withOplog || opts.newOplogMsg != nil) { @@ -531,7 +531,7 @@ func (rw *Db) Update(ctx context.Context, i interface{}, fieldMaskPaths []string // from the db opt = append(opt, WithLookup(true)) if err := rw.lookupAfterWrite(ctx, i, opt...); err != nil { - return NoRowsAffected, errors.Wrap(ctx, err, op) + return NoRowsAffected, errors.Wrap(ctx, err, op, errors.WithoutEvent()) } return rowsUpdated, nil } @@ -582,7 +582,7 @@ func (rw *Db) Delete(ctx context.Context, i interface{}, opt ...Option) (int, er } db = db.Delete(i) if db.Error != nil { - return NoRowsAffected, errors.Wrap(ctx, db.Error, op) + return NoRowsAffected, errors.Wrap(ctx, db.Error, op, errors.WithoutEvent()) } rowsDeleted := int(db.RowsAffected) if rowsDeleted > 0 && (withOplog || opts.newOplogMsg != nil) { @@ -650,7 +650,7 @@ func (rw *Db) DeleteItems(ctx context.Context, deleteItems []interface{}, opt .. // relationship between Create and CreateItems). underlying := rw.underlying.Delete(item) if underlying.Error != nil { - return rowsDeleted, errors.Wrap(ctx, underlying.Error, op) + return rowsDeleted, errors.Wrap(ctx, underlying.Error, op, errors.WithoutEvent()) } rowsDeleted += int(underlying.RowsAffected) } @@ -675,14 +675,14 @@ func validateOplogArgs(ctx context.Context, i interface{}, opts Options) (oplog. const op = "db.validateOplogArgs" oplogArgs := opts.oplogOpts if oplogArgs.wrapper == nil { - return nil, errors.New(ctx, errors.InvalidParameter, op, "missing wrapper") + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing wrapper", errors.WithoutEvent()) } if len(oplogArgs.metadata) == 0 { - return nil, errors.New(ctx, errors.InvalidParameter, op, "missing metadata") + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing metadata", errors.WithoutEvent()) } replayable, ok := i.(oplog.ReplayableMessage) if !ok { - return nil, errors.E(ctx, errors.WithOp(op), errors.WithMsg("not a replayable message")) + return nil, errors.E(ctx, errors.WithOp(op), errors.WithMsg("not a replayable message"), errors.WithoutEvent()) } return replayable, nil } @@ -690,15 +690,15 @@ func validateOplogArgs(ctx context.Context, i interface{}, opts Options) (oplog. func (rw *Db) getTicketFor(aggregateName string) (*store.Ticket, error) { const op = "db.getTicketFor" if rw.underlying == nil { - return nil, errors.NewDeprecated(errors.InvalidParameter, op, fmt.Sprintf("%s: underlying db missing", aggregateName)) + return nil, errors.NewDeprecated(errors.InvalidParameter, op, fmt.Sprintf("%s: underlying db missing", aggregateName), errors.WithoutEvent()) } ticketer, err := oplog.NewGormTicketer(rw.underlying, oplog.WithAggregateNames(true)) if err != nil { - return nil, errors.WrapDeprecated(err, op, errors.WithMsg(fmt.Sprintf("%s: unable to get Ticketer", aggregateName))) + return nil, errors.WrapDeprecated(err, op, errors.WithMsg(fmt.Sprintf("%s: unable to get Ticketer", aggregateName)), errors.WithoutEvent()) } ticket, err := ticketer.GetTicket(aggregateName) if err != nil { - return nil, errors.WrapDeprecated(err, op, errors.WithMsg(fmt.Sprintf("%s: unable to get ticket", aggregateName))) + return nil, errors.WrapDeprecated(err, op, errors.WithMsg(fmt.Sprintf("%s: unable to get ticket", aggregateName)), errors.WithoutEvent()) } return ticket, nil } @@ -708,14 +708,14 @@ func (rw *Db) getTicketFor(aggregateName string) (*store.Ticket, error) { func (rw *Db) GetTicket(i interface{}) (*store.Ticket, error) { const op = "db.GetTicket" if rw.underlying == nil { - return nil, errors.NewDeprecated(errors.InvalidParameter, op, "missing underlying db") + return nil, errors.NewDeprecated(errors.InvalidParameter, op, "missing underlying db", errors.WithoutEvent()) } if isNil(i) { - return nil, errors.NewDeprecated(errors.InvalidParameter, op, "missing interface") + return nil, errors.NewDeprecated(errors.InvalidParameter, op, "missing interface", errors.WithoutEvent()) } replayable, ok := i.(oplog.ReplayableMessage) if !ok { - return nil, errors.NewDeprecated(errors.InvalidParameter, op, "not a replayable message") + return nil, errors.NewDeprecated(errors.InvalidParameter, op, "not a replayable message", errors.WithoutEvent()) } return rw.getTicketFor(replayable.TableName()) } @@ -723,7 +723,7 @@ func (rw *Db) GetTicket(i interface{}) (*store.Ticket, error) { func (rw *Db) oplogMsgsForItems(ctx context.Context, opType OpType, opts Options, items []interface{}) ([]*oplog.Message, error) { const op = "db.oplogMsgsForItems" if len(items) == 0 { - return nil, errors.New(ctx, errors.InvalidParameter, op, "missing items") + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing items", errors.WithoutEvent()) } oplogMsgs := []*oplog.Message{} var foundType reflect.Type @@ -733,11 +733,11 @@ func (rw *Db) oplogMsgsForItems(ctx context.Context, opType OpType, opts Options } currentType := reflect.TypeOf(item) if foundType != currentType { - return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("items contains disparate types. item (%d) %s is not a %s", i, currentType, foundType)) + return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("items contains disparate types. item (%d) %s is not a %s", i, currentType, foundType), errors.WithoutEvent()) } msg, err := rw.newOplogMessage(ctx, opType, item, WithFieldMaskPaths(opts.WithFieldMaskPaths), WithNullPaths(opts.WithNullPaths)) if err != nil { - return nil, errors.Wrap(ctx, err, op) + return nil, errors.Wrap(ctx, err, op, errors.WithoutEvent()) } oplogMsgs = append(oplogMsgs, msg) } @@ -751,16 +751,16 @@ func (rw *Db) addOplogForItems(ctx context.Context, opType OpType, opts Options, const op = "db.addOplogForItems" oplogArgs := opts.oplogOpts if ticket == nil { - return errors.New(ctx, errors.InvalidParameter, op, "missing ticket") + return errors.New(ctx, errors.InvalidParameter, op, "missing ticket", errors.WithoutEvent()) } if len(items) == 0 { - return errors.New(ctx, errors.InvalidParameter, op, "missing items") + return errors.New(ctx, errors.InvalidParameter, op, "missing items", errors.WithoutEvent()) } if oplogArgs.metadata == nil { - return errors.New(ctx, errors.InvalidParameter, op, "missing metadata") + return errors.New(ctx, errors.InvalidParameter, op, "missing metadata", errors.WithoutEvent()) } if oplogArgs.wrapper == nil { - return errors.New(ctx, errors.InvalidParameter, op, "missing wrapper") + return errors.New(ctx, errors.InvalidParameter, op, "missing wrapper", errors.WithoutEvent()) } oplogMsgs, err := rw.oplogMsgsForItems(ctx, opType, opts, items) @@ -770,11 +770,11 @@ func (rw *Db) addOplogForItems(ctx context.Context, opType OpType, opts Options, replayable, err := validateOplogArgs(ctx, items[0], opts) if err != nil { - return errors.Wrap(ctx, err, op, errors.WithMsg("oplog validation failed")) + return errors.Wrap(ctx, err, op, errors.WithMsg("oplog validation failed"), errors.WithoutEvent()) } ticketer, err := oplog.NewGormTicketer(rw.underlying, oplog.WithAggregateNames(true)) if err != nil { - return errors.Wrap(ctx, err, op, errors.WithMsg("unable to get Ticketer")) + return errors.Wrap(ctx, err, op, errors.WithMsg("unable to get Ticketer"), errors.WithoutEvent()) } entry, err := oplog.NewEntry( replayable.TableName(), @@ -783,7 +783,7 @@ func (rw *Db) addOplogForItems(ctx context.Context, opType OpType, opts Options, ticketer, ) if err != nil { - return errors.Wrap(ctx, err, op, errors.WithMsg("unable to create oplog entry")) + return errors.Wrap(ctx, err, op, errors.WithMsg("unable to create oplog entry"), errors.WithoutEvent()) } if err := entry.WriteEntryWith( ctx, @@ -791,7 +791,7 @@ func (rw *Db) addOplogForItems(ctx context.Context, opType OpType, opts Options, ticket, oplogMsgs..., ); err != nil { - return errors.Wrap(ctx, err, op, errors.WithMsg("unable to write oplog entry")) + return errors.Wrap(ctx, err, op, errors.WithMsg("unable to write oplog entry"), errors.WithoutEvent()) } return nil } @@ -801,14 +801,14 @@ func (rw *Db) addOplog(ctx context.Context, opType OpType, opts Options, ticket oplogArgs := opts.oplogOpts replayable, err := validateOplogArgs(ctx, i, opts) if err != nil { - return errors.Wrap(ctx, err, op) + return errors.Wrap(ctx, err, op, errors.WithoutEvent()) } if ticket == nil { - return errors.New(ctx, errors.InvalidParameter, op, "missing ticket") + return errors.New(ctx, errors.InvalidParameter, op, "missing ticket", errors.WithoutEvent()) } ticketer, err := oplog.NewGormTicketer(rw.underlying, oplog.WithAggregateNames(true)) if err != nil { - return errors.Wrap(ctx, err, op, errors.WithMsg("unable to get Ticketer")) + return errors.Wrap(ctx, err, op, errors.WithMsg("unable to get Ticketer"), errors.WithoutEvent()) } entry, err := oplog.NewEntry( replayable.TableName(), @@ -817,11 +817,11 @@ func (rw *Db) addOplog(ctx context.Context, opType OpType, opts Options, ticket ticketer, ) if err != nil { - return errors.Wrap(ctx, err, op) + return errors.Wrap(ctx, err, op, errors.WithoutEvent()) } msg, err := rw.newOplogMessage(ctx, opType, i, WithFieldMaskPaths(opts.WithFieldMaskPaths), WithNullPaths(opts.WithNullPaths)) if err != nil { - return errors.Wrap(ctx, err, op) + return errors.Wrap(ctx, err, op, errors.WithoutEvent()) } err = entry.WriteEntryWith( ctx, @@ -830,7 +830,7 @@ func (rw *Db) addOplog(ctx context.Context, opType OpType, opts Options, ticket msg, ) if err != nil { - return errors.Wrap(ctx, err, op, errors.WithMsg("unable to write oplog entry")) + return errors.Wrap(ctx, err, op, errors.WithMsg("unable to write oplog entry"), errors.WithoutEvent()) } return nil } @@ -886,7 +886,7 @@ func (rw *Db) newOplogMessage(ctx context.Context, opType OpType, i interface{}, opts := GetOpts(opt...) replayable, ok := i.(oplog.ReplayableMessage) if !ok { - return nil, errors.New(ctx, errors.InvalidParameter, op, "not a replayable interface") + return nil, errors.New(ctx, errors.InvalidParameter, op, "not a replayable interface", errors.WithoutEvent()) } msg := oplog.Message{ Message: i.(proto.Message), @@ -902,7 +902,7 @@ func (rw *Db) newOplogMessage(ctx context.Context, opType OpType, i interface{}, case DeleteOp: msg.OpType = oplog.OpType_OP_TYPE_DELETE default: - return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("operation type %v is not supported", opType)) + return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("operation type %v is not supported", opType), errors.WithoutEvent()) } return &msg, nil } @@ -919,7 +919,7 @@ func (w *Db) DoTx(ctx context.Context, retries uint, backOff Backoff, Handler Tx info := RetryInfo{} for attempts := uint(1); ; attempts++ { if attempts > retries+1 { - return info, errors.New(ctx, errors.MaxRetries, op, fmt.Sprintf("Too many retries: %d of %d", attempts-1, retries+1)) + return info, errors.New(ctx, errors.MaxRetries, op, fmt.Sprintf("Too many retries: %d of %d", attempts-1, retries+1), errors.WithoutEvent()) } // step one of this, start a transaction... @@ -928,7 +928,7 @@ func (w *Db) DoTx(ctx context.Context, retries uint, backOff Backoff, Handler Tx rw := &Db{newTx} if err := Handler(rw, rw); err != nil { if err := newTx.Rollback().Error; err != nil { - return info, errors.Wrap(ctx, err, op) + return info, errors.Wrap(ctx, err, op, errors.WithoutEvent()) } if errors.Match(errors.T(errors.TicketAlreadyRedeemed), err) { d := backOff.Duration(attempts) @@ -966,9 +966,9 @@ func (rw *Db) LookupById(ctx context.Context, resourceWithIder interface{}, _ .. } if err := rw.underlying.Where(where, primaryKey).First(resourceWithIder).Error; err != nil { if err == gorm.ErrRecordNotFound { - return errors.E(ctx, errors.WithCode(errors.RecordNotFound), errors.WithOp(op)) + return errors.E(ctx, errors.WithCode(errors.RecordNotFound), errors.WithOp(op), errors.WithoutEvent()) } - return errors.Wrap(ctx, err, op) + return errors.Wrap(ctx, err, op, errors.WithoutEvent()) } return nil } @@ -984,10 +984,10 @@ func primaryKeyWhere(ctx context.Context, resourceWithIder interface{}) (pkey st primaryKey = resourceType.GetPrivateId() where = "private_id = ?" default: - return "", "", errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("unsupported interface type %T", resourceWithIder)) + return "", "", errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("unsupported interface type %T", resourceWithIder), errors.WithoutEvent()) } if primaryKey == "" { - return "", "", errors.New(ctx, errors.InvalidParameter, op, "missing primary key") + return "", "", errors.New(ctx, errors.InvalidParameter, op, "missing primary key", errors.WithoutEvent()) } return primaryKey, where, nil } @@ -1009,9 +1009,9 @@ func (rw *Db) LookupWhere(ctx context.Context, resource interface{}, where strin } if err := rw.underlying.Where(where, args...).First(resource).Error; err != nil { if err == gorm.ErrRecordNotFound { - return errors.E(ctx, errors.WithCode(errors.RecordNotFound), errors.WithOp(op)) + return errors.E(ctx, errors.WithCode(errors.RecordNotFound), errors.WithOp(op), errors.WithoutEvent()) } - return errors.Wrap(ctx, err, op) + return errors.Wrap(ctx, err, op, errors.WithoutEvent()) } return nil } @@ -1055,7 +1055,7 @@ func (rw *Db) SearchWhere(ctx context.Context, resources interface{}, where stri err = db.Find(resources).Error if err != nil { // searching with a slice parameter does not return a gorm.ErrRecordNotFound - return errors.Wrap(ctx, err, op) + return errors.Wrap(ctx, err, op, errors.WithoutEvent()) } return nil } diff --git a/internal/errors/error.go b/internal/errors/error.go index 7944381cc..73f0f396a 100644 --- a/internal/errors/error.go +++ b/internal/errors/error.go @@ -107,7 +107,7 @@ func E(ctx context.Context, opt ...Option) error { eventErr.Op = "unknown operation" } } - event.WriteError(ctx, event.Op(err.Op), eventErr) + event.WriteError(ctx, event.Op(eventErr.Op), eventErr) } return err @@ -182,6 +182,7 @@ func WrapDeprecated(e error, op Op, opt ...Option) error { // and attempt to add a helpful error msg as well. If that's not possible, it // will return nil func Convert(e error) *Err { + ctx := context.TODO() if e == nil { return nil } @@ -196,24 +197,24 @@ func Convert(e error) *Err { if pqError.Code.Class() == "23" { // class of integrity constraint violations switch pqError.Code { case "23505": // unique_violation - return E(context.TODO(), WithoutEvent(), WithMsg(pqError.Message), WithWrap(EDeprecated(WithCode(NotUnique), WithMsg("unique constraint violation")))).(*Err) + return E(ctx, WithoutEvent(), WithMsg(pqError.Message), WithWrap(E(ctx, WithoutEvent(), WithCode(NotUnique), WithMsg("unique constraint violation")))).(*Err) case "23502": // not_null_violation msg := fmt.Sprintf("%s must not be empty", pqError.Column) - return E(context.TODO(), WithoutEvent(), WithMsg(msg), WithWrap(EDeprecated(WithCode(NotNull), WithMsg("not null constraint violated")))).(*Err) + return E(ctx, WithoutEvent(), WithMsg(msg), WithWrap(E(ctx, WithoutEvent(), WithCode(NotNull), WithMsg("not null constraint violated")))).(*Err) case "23514": // check_violation msg := fmt.Sprintf("%s constraint failed", pqError.Constraint) - return E(context.TODO(), WithoutEvent(), WithMsg(msg), WithWrap(EDeprecated(WithCode(CheckConstraint), WithMsg("check constraint violated")))).(*Err) + return E(ctx, WithoutEvent(), WithMsg(msg), WithWrap(E(ctx, WithoutEvent(), WithCode(CheckConstraint), WithMsg("check constraint violated")))).(*Err) default: - return E(context.TODO(), WithoutEvent(), WithCode(NotSpecificIntegrity), WithMsg(pqError.Message)).(*Err) + return E(ctx, WithoutEvent(), WithCode(NotSpecificIntegrity), WithMsg(pqError.Message)).(*Err) } } switch pqError.Code { case "42P01": - return E(context.TODO(), WithoutEvent(), WithCode(MissingTable), WithMsg(pqError.Message)).(*Err) + return E(ctx, WithoutEvent(), WithCode(MissingTable), WithMsg(pqError.Message)).(*Err) case "42703": - return E(context.TODO(), WithoutEvent(), WithCode(ColumnNotFound), WithMsg(pqError.Message)).(*Err) + return E(ctx, WithoutEvent(), WithCode(ColumnNotFound), WithMsg(pqError.Message)).(*Err) case "P0001": - return E(context.TODO(), WithoutEvent(), WithCode(Exception), WithMsg(pqError.Message)).(*Err) + return E(ctx, WithoutEvent(), WithCode(Exception), WithMsg(pqError.Message)).(*Err) } } // unfortunately, we can't help. diff --git a/internal/scheduler/job/repository_job.go b/internal/scheduler/job/repository_job.go index 7e56077bc..73aceb2f8 100644 --- a/internal/scheduler/job/repository_job.go +++ b/internal/scheduler/job/repository_job.go @@ -38,24 +38,24 @@ func (r *Repository) CreateJob(ctx context.Context, name, description string, op int(opts.withNextRunIn.Round(time.Second).Seconds()), }) if err != nil { - return errors.Wrap(ctx, err, op) + return errors.Wrap(ctx, err, op, errors.WithoutEvent()) } defer rows.Close() var rowCnt int for rows.Next() { if rowCnt > 0 { - return errors.New(ctx, errors.MultipleRecords, op, "more than 1 job would have been created") + return errors.New(ctx, errors.MultipleRecords, op, "more than 1 job would have been created", errors.WithoutEvent()) } rowCnt++ err = r.ScanRows(rows, j) if err != nil { _ = rows.Close() - return errors.Wrap(ctx, err, op, errors.WithMsg("unable to scan rows for job")) + return errors.Wrap(ctx, err, op, errors.WithMsg("unable to scan rows for job"), errors.WithoutEvent()) } } if rowCnt == 0 { - return errors.New(ctx, errors.NotSpecificIntegrity, op, "failed to create new job") + return errors.New(ctx, errors.NotSpecificIntegrity, op, "failed to create new job", errors.WithoutEvent()) } return nil @@ -63,7 +63,7 @@ func (r *Repository) CreateJob(ctx context.Context, name, description string, op ) if err != nil { if errors.IsUniqueError(err) { - return nil, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("name %s already exists", name))) + return nil, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("name %s already exists", name)), errors.WithoutEvent()) } return nil, errors.Wrap(ctx, err, op) } diff --git a/internal/servers/controller/handlers/workers/worker_service.go b/internal/servers/controller/handlers/workers/worker_service.go index c228525dc..cb6a3f2f5 100644 --- a/internal/servers/controller/handlers/workers/worker_service.go +++ b/internal/servers/controller/handlers/workers/worker_service.go @@ -215,7 +215,6 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques func (ws *workerServiceServer) LookupSession(ctx context.Context, req *pbs.LookupSessionRequest) (*pbs.LookupSessionResponse, error) { const op = "workers.(workerServiceServer).LookupSession" - event.WriteSysEvent(ctx, op, "got validate session request from worker", "session_id", req.GetSessionId()) sessRepo, err := ws.sessionRepoFn() if err != nil { @@ -316,7 +315,6 @@ func (ws *workerServiceServer) LookupSession(ctx context.Context, req *pbs.Looku func (ws *workerServiceServer) CancelSession(ctx context.Context, req *pbs.CancelSessionRequest) (*pbs.CancelSessionResponse, error) { const op = "workers.(workerServiceServer).CancelSession" - event.WriteSysEvent(ctx, op, "got cancel session request from worker", "session_id", req.GetSessionId()) sessRepo, err := ws.sessionRepoFn() if err != nil { @@ -340,7 +338,6 @@ func (ws *workerServiceServer) CancelSession(ctx context.Context, req *pbs.Cance func (ws *workerServiceServer) ActivateSession(ctx context.Context, req *pbs.ActivateSessionRequest) (*pbs.ActivateSessionResponse, error) { const op = "workers.(workerServiceServer).ActivateSession" - event.WriteSysEvent(ctx, op, "got activate session request from worker", "session_id", req.GetSessionId()) sessRepo, err := ws.sessionRepoFn() if err != nil { @@ -364,14 +361,6 @@ func (ws *workerServiceServer) ActivateSession(ctx context.Context, req *pbs.Act return nil, status.Error(codes.Internal, "Invalid session state in activate response.") } - event.WriteSysEvent(ctx, op, "session activated", - "session_id", sessionInfo.PublicId, - "target_id", sessionInfo.TargetId, - "user_id", sessionInfo.UserId, - "host_set_id", sessionInfo.HostSetId, - "host_id", sessionInfo.HostId, - ) - return &pbs.ActivateSessionResponse{ Status: sessionStates[0].Status.ProtoVal(), }, nil @@ -379,7 +368,6 @@ func (ws *workerServiceServer) ActivateSession(ctx context.Context, req *pbs.Act func (ws *workerServiceServer) AuthorizeConnection(ctx context.Context, req *pbs.AuthorizeConnectionRequest) (*pbs.AuthorizeConnectionResponse, error) { const op = "workers.(workerServiceServer" - event.WriteSysEvent(ctx, op, "got authorize connection request from worker", "session_id", req.GetSessionId()) sessRepo, err := ws.sessionRepoFn() if err != nil { @@ -406,18 +394,11 @@ func (ws *workerServiceServer) AuthorizeConnection(ctx context.Context, req *pbs ret.ConnectionsLeft -= int32(authzSummary.CurrentConnectionCount) } - event.WriteSysEvent(ctx, op, "authorized connection", - "session_id", req.GetSessionId(), - "connection_id", ret.ConnectionId, - "connections_left", ret.ConnectionsLeft, - ) - return ret, nil } func (ws *workerServiceServer) ConnectConnection(ctx context.Context, req *pbs.ConnectConnectionRequest) (*pbs.ConnectConnectionResponse, error) { const op = "workers.(workerServiceServer).ConnectConnection" - event.WriteSysEvent(ctx, op, "got connection established information from worker", "connection_id", req.GetConnectionId()) sessRepo, err := ws.sessionRepoFn() if err != nil { @@ -456,8 +437,6 @@ func (ws *workerServiceServer) ConnectConnection(ctx context.Context, req *pbs.C ) } - event.WriteSysEvent(ctx, op, "connection established", "logger_pairs", loggerPairs) - return ret, nil } @@ -480,8 +459,6 @@ func (ws *workerServiceServer) CloseConnection(ctx context.Context, req *pbs.Clo ClosedReason: session.ClosedReason(v.GetReason()), }) } - event.WriteSysEvent(ctx, op, "got connection close information from worker", "connection_ids", closeIds) - sessRepo, err := ws.sessionRepoFn() if err != nil { return nil, status.Errorf(codes.Internal, "error getting session repo: %v", err) @@ -509,10 +486,6 @@ func (ws *workerServiceServer) CloseConnection(ctx context.Context, req *pbs.Clo }) } - for _, v := range req.GetCloseRequestData() { - event.WriteSysEvent(ctx, op, "connection closed", "connection_id", v.ConnectionId) - } - ret := &pbs.CloseConnectionResponse{ CloseResponseData: closeData, } diff --git a/internal/servers/controller/tickers.go b/internal/servers/controller/tickers.go index f3873adb2..84ca7cdc9 100644 --- a/internal/servers/controller/tickers.go +++ b/internal/servers/controller/tickers.go @@ -106,7 +106,6 @@ func (c *Controller) startTerminateCompletedSessionsTicking(cancelCtx context.Co if err != nil { event.WriteError(cancelCtx, op, err, event.WithInfoMsg("error performing termination of completed sessions")) } else if terminationCount > 0 { - event.WriteSysEvent(cancelCtx, op, "terminating completed sessions successful", "sessions_terminated", terminationCount) } } timer.Reset(getRandomInterval()) diff --git a/internal/servers/worker/handler.go b/internal/servers/worker/handler.go index 91b610985..2c22faf35 100644 --- a/internal/servers/worker/handler.go +++ b/internal/servers/worker/handler.go @@ -61,8 +61,6 @@ func (w *Worker) handleProxy() http.HandlerFunc { Port: numPort, } - event.WriteSysEvent(ctx, op, "received TLS connection") - siRaw, valid := w.sessionInfoMap.Load(sessionId) if !valid { event.WriteError(ctx, op, errors.New("session not found in info map"), event.WithInfo("session_id", sessionId)) @@ -79,8 +77,6 @@ func (w *Worker) handleProxy() http.HandlerFunc { sessStatus := si.status si.RUnlock() - event.WriteSysEvent(ctx, op, "found session in session info map") - opts := &websocket.AcceptOptions{ Subprotocols: []string{globals.TcpProxyV1}, } @@ -93,8 +89,6 @@ func (w *Worker) handleProxy() http.HandlerFunc { // Later calls will cause this to noop if they return a different status defer conn.Close(websocket.StatusNormalClosure, "done") - event.WriteSysEvent(ctx, op, "websocket upgrade done") - connCtx, connCancel := context.WithDeadline(r.Context(), expiration.AsTime()) defer connCancel() @@ -110,8 +104,6 @@ func (w *Worker) handleProxy() http.HandlerFunc { return } - event.WriteSysEvent(ctx, op, "proxy handshake finished") - if tofuToken != "" { if tofuToken != handshake.GetTofuToken() { event.WriteError(ctx, op, errors.New("WARNING: mismatched tofu token"), event.WithInfo("session_id", sessionId)) @@ -125,7 +117,6 @@ func (w *Worker) handleProxy() http.HandlerFunc { return } if handshake.Command == proxy.HANDSHAKECOMMAND_HANDSHAKECOMMAND_UNSPECIFIED { - event.WriteSysEvent(ctx, op, "activating session") sessStatus, err = w.activateSession(r.Context(), sessionId, handshake.GetTofuToken(), version) if err != nil { event.WriteError(ctx, op, err, event.WithInfoMsg("unable to validate session")) @@ -136,7 +127,6 @@ func (w *Worker) handleProxy() http.HandlerFunc { } if handshake.Command == proxy.HANDSHAKECOMMAND_HANDSHAKECOMMAND_SESSION_CANCEL { - event.WriteSysEvent(ctx, op, "canceling session at client request") _, err := w.cancelSession(r.Context(), sessionId) if err != nil { event.WriteError(ctx, op, err, event.WithInfoMsg("unable to cancel session")) @@ -168,8 +158,6 @@ func (w *Worker) handleProxy() http.HandlerFunc { connectionLimit := si.lookupSessionResponse.GetConnectionLimit() si.Unlock() - event.WriteSysEvent(ctx, op, "authorized connection", "connection_id", ci.id) - handshakeResult := &proxy.HandshakeResult{ Expiration: expiration, ConnectionLimit: connectionLimit, diff --git a/internal/servers/worker/session.go b/internal/servers/worker/session.go index a2b3e63e7..9c5da6bc3 100644 --- a/internal/servers/worker/session.go +++ b/internal/servers/worker/session.go @@ -43,7 +43,6 @@ func (w *Worker) getSessionTls(hello *tls.ClientHelloInfo) (*tls.Config, error) var sessionId string switch { case strings.HasPrefix(hello.ServerName, "s_"): - event.WriteSysEvent(ctx, op, "got valid session in SNI", "session_id", hello.ServerName) sessionId = hello.ServerName default: event.WriteSysEvent(ctx, op, "invalid session in SNI", "session_id", hello.ServerName) @@ -68,7 +67,6 @@ func (w *Worker) getSessionTls(hello *tls.ClientHelloInfo) (*tls.Config, error) timeoutContext, cancel := context.WithTimeout(w.baseContext, validateSessionTimeout) defer cancel() - event.WriteSysEvent(ctx, op, "looking up session", "session_id", sessionId) resp, err := conn.LookupSession(timeoutContext, &pbs.LookupSessionRequest{ ServerId: w.conf.RawConfig.Worker.Name, SessionId: sessionId, @@ -127,7 +125,6 @@ func (w *Worker) getSessionTls(hello *tls.ClientHelloInfo) (*tls.Config, error) actualSi.Unlock() } - event.WriteSysEvent(ctx, op, "returning TLS configuration", "session_id", sessionId) return tlsConf, nil } @@ -270,8 +267,6 @@ func (w *Worker) closeConnections(ctx context.Context, closeInfo map[string]stri return } - event.WriteSysEvent(ctx, op, "marking connections as closed", "session_and_connection_ids", fmt.Sprintf("%#v", closeInfo)) - // How we handle close info depends on whether or not we succeeded with // marking them closed on the controller. var sessionCloseInfo map[string][]*pbs.CloseConnectionResponseData @@ -304,14 +299,13 @@ func (w *Worker) closeConnections(ctx context.Context, closeInfo map[string]stri } // Mark connections as closed - closedIds, errs := w.setCloseTimeForResponse(sessionCloseInfo) + _, errs := w.setCloseTimeForResponse(sessionCloseInfo) if len(errs) > 0 { for _, err := range errs { event.WriteError(ctx, op, err, event.WithInfoMsg("error marking connection closed in state")) } } - event.WriteSysEvent(ctx, op, "connections successfully marked closed", "connection_ids", closedIds) } // makeCloseConnectionRequest creates a CloseConnectionRequest for diff --git a/internal/servers/worker/status.go b/internal/servers/worker/status.go index 5ab1a77a0..d96b44cfd 100644 --- a/internal/servers/worker/status.go +++ b/internal/servers/worker/status.go @@ -188,7 +188,6 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context) { w.lastStatusSuccess.Store(&LastStatusInformation{StatusResponse: result, StatusTime: time.Now()}) for _, request := range result.GetJobsRequests() { - event.WriteSysEvent(statusCtx, op, "got job request from controller", "request", request) switch request.GetRequestType() { case pbs.CHANGETYPE_CHANGETYPE_UPDATE_STATE: switch request.GetJob().GetType() { diff --git a/internal/servers/worker/tcp_proxy.go b/internal/servers/worker/tcp_proxy.go index 66b210464..328ef944a 100644 --- a/internal/servers/worker/tcp_proxy.go +++ b/internal/servers/worker/tcp_proxy.go @@ -67,17 +67,15 @@ func (w *Worker) handleTcpProxyV1(connCtx context.Context, clientAddr *net.TCPAd connWg.Add(2) go func() { defer connWg.Done() - _, err := io.Copy(netConn, tcpRemoteConn) + _, _ = io.Copy(netConn, tcpRemoteConn) netConn.Close() tcpRemoteConn.Close() - event.WriteSysEvent(ctx, op, "copy from client to endpoint done", "error", err) }() go func() { defer connWg.Done() - _, err := io.Copy(tcpRemoteConn, netConn) + _, _ = io.Copy(tcpRemoteConn, netConn) tcpRemoteConn.Close() netConn.Close() - event.WriteSysEvent(ctx, op, "copy from endpoint to client done", "error", err) }() connWg.Wait() }