CyberCoveLLC/threshold
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Events: ongoing clean up of eventing (#1448)

Browse Source
This commit is contained in:
Jim authored and GitHub committed 2021-08-05 10:54:10 -04:00
1 parent 00cb18efbe
commit 477b541bf0
9 files changed
+65 -113
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/db/read_writer.go
+48 -48
View File
@@ -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
}
internal/errors/error.go
+9 -8
View File
@@ -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.
internal/scheduler/job/repository_job.go
+5 -5
View File
@@ -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)
}
internal/servers/controller/handlers/workers/worker_service.go
-27
View File
@@ -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,
}
internal/servers/controller/tickers.go
-1
View File
@@ -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())
internal/servers/worker/handler.go
-12
View File
@@ -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,
internal/servers/worker/session.go
+1 -7
View File
@@ -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
internal/servers/worker/status.go
-1
View File
@@ -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() {
internal/servers/worker/tcp_proxy.go
+2 -4
View File
@@ -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()
}