diff --git a/internal/db/schema/migrations/oss/postgres/20/01_session_list_view.up.sql b/internal/db/schema/migrations/oss/postgres/20/01_session_list_view.up.sql new file mode 100644 index 000000000..2feb2a68c --- /dev/null +++ b/internal/db/schema/migrations/oss/postgres/20/01_session_list_view.up.sql @@ -0,0 +1,50 @@ +begin; + +-- Replaces the view created in 1/01 to include connections +drop view session_with_state; +create view session_list as + select + s.public_id, + s.user_id, + s.host_id, + s.server_id, + s.server_type, + s.target_id, + s.host_set_id, + s.auth_token_id, + s.scope_id, + s.certificate, + s.expiration_time, + s.connection_limit, + s.tofu_token, + s.key_id, + s.termination_reason, + s.version, + s.create_time, + s.update_time, + s.endpoint, + s.worker_filter, + ss.state, + ss.previous_end_time, + ss.start_time, + ss.end_time, + sc.public_id as connection_id, + sc.client_tcp_address, + sc.client_tcp_port, + sc.endpoint_tcp_address, + sc.endpoint_tcp_port, + sc.bytes_up, + sc.bytes_down, + sc.closed_reason + from + session s + join + session_state ss + on + s.public_id = ss.session_id + left join + session_connection sc + on + s.public_id = sc.session_id; + +commit; diff --git a/internal/session/query.go b/internal/session/query.go index 8218e43f5..0da27a89a 100644 --- a/internal/session/query.go +++ b/internal/session/query.go @@ -155,7 +155,7 @@ from select * from (select public_id from session %s) s, - session_with_state ss + session_list ss where s.public_id = ss.public_id %s diff --git a/internal/session/repository.go b/internal/session/repository.go index b4c75ec1e..8735c8d7f 100644 --- a/internal/session/repository.go +++ b/internal/session/repository.go @@ -5,6 +5,7 @@ import ( "sort" "github.com/hashicorp/boundary/internal/db" + "github.com/hashicorp/boundary/internal/db/timestamp" "github.com/hashicorp/boundary/internal/errors" "github.com/hashicorp/boundary/internal/kms" ) @@ -74,19 +75,31 @@ func (r *Repository) list(ctx context.Context, resources interface{}, where stri return nil } -func (r *Repository) convertToSessions(ctx context.Context, sessionsWithState []*sessionView, opt ...Option) ([]*Session, error) { +func (r *Repository) convertToSessions(ctx context.Context, sessionList []*sessionListView, opt ...Option) ([]*Session, error) { const op = "session.(Repository).convertToSessions" opts := getOpts(opt...) - if len(sessionsWithState) == 0 { + if len(sessionList) == 0 { return nil, nil } sessions := []*Session{} + // deduplication map of connections by connection_id + connections := map[string]*Connection{} + // deduplication map of states by end_time + states := map[*timestamp.Timestamp]*State{} var prevSessionId string var workingSession *Session - for _, sv := range sessionsWithState { + for _, sv := range sessionList { if sv.PublicId != prevSessionId { if prevSessionId != "" { + for _, c := range connections { + workingSession.Connections = append(workingSession.Connections, c) + } + connections = map[string]*Connection{} + for _, s := range states { + workingSession.States = append(workingSession.States, s) + } + states = map[*timestamp.Timestamp]*State{} sort.Slice(workingSession.States, func(i, j int) bool { return workingSession.States[i].StartTime.GetTimestamp().AsTime().After(workingSession.States[j].StartTime.GetTimestamp().AsTime()) }) @@ -134,18 +147,42 @@ func (r *Repository) convertToSessions(ctx context.Context, sessionsWithState [] } } - state := &State{ - SessionId: sv.PublicId, - Status: Status(sv.Status), - PreviousEndTime: sv.PreviousEndTime, - StartTime: sv.StartTime, - EndTime: sv.EndTime, + if _, ok := states[sv.EndTime]; !ok { + states[sv.EndTime] = &State{ + SessionId: sv.PublicId, + Status: Status(sv.Status), + PreviousEndTime: sv.PreviousEndTime, + StartTime: sv.StartTime, + EndTime: sv.EndTime, + } } - workingSession.States = append(workingSession.States, state) + + if sv.ConnectionId != "" { + if _, ok := connections[sv.ConnectionId]; !ok { + connections[sv.ConnectionId] = &Connection{ + PublicId: sv.ConnectionId, + SessionId: sv.PublicId, + ClientTcpAddress: sv.ClientTcpAddress, + ClientTcpPort: sv.ClientTcpPort, + EndpointTcpAddress: sv.EndpointTcpAddress, + EndpointTcpPort: sv.EndpointTcpPort, + BytesUp: sv.BytesUp, + BytesDown: sv.BytesDown, + ClosedReason: sv.ClosedReason, + } + } + } + + } + for _, s := range states { + workingSession.States = append(workingSession.States, s) } sort.Slice(workingSession.States, func(i, j int) bool { return workingSession.States[i].StartTime.GetTimestamp().AsTime().After(workingSession.States[j].StartTime.GetTimestamp().AsTime()) }) + for _, c := range connections { + workingSession.Connections = append(workingSession.Connections, c) + } sessions = append(sessions, workingSession) return sessions, nil } diff --git a/internal/session/repository_session.go b/internal/session/repository_session.go index f324be3b6..7e14cae7f 100644 --- a/internal/session/repository_session.go +++ b/internal/session/repository_session.go @@ -166,6 +166,12 @@ func (r *Repository) LookupSession(ctx context.Context, sessionId string, _ ...O if len(creds) > 0 { session.DynamicCredentials = creds } + + connections, err := fetchConnections(ctx, read, sessionId, db.WithOrder("create_time desc")) + if err != nil { + return errors.Wrap(ctx, err, op) + } + session.Connections = connections return nil }, ) @@ -249,6 +255,8 @@ func (r *Repository) ListSessions(ctx context.Context, opt ...Option) ([]*Sessio case db.AscendingOrderBy: withOrder = "order by create_time asc" case db.DescendingOrderBy: + fallthrough + default: withOrder = "order by create_time" } @@ -265,15 +273,15 @@ func (r *Repository) ListSessions(ctx context.Context, opt ...Option) ([]*Sessio } defer rows.Close() - var sessionsWithState []*sessionView + var sessionsList []*sessionListView for rows.Next() { - var s sessionView + var s sessionListView if err := r.reader.ScanRows(rows, &s); err != nil { return nil, errors.Wrap(ctx, err, op, errors.WithMsg("scan row failed")) } - sessionsWithState = append(sessionsWithState, &s) + sessionsList = append(sessionsList, &s) } - sessions, err := r.convertToSessions(ctx, sessionsWithState, withListingConvert(true)) + sessions, err := r.convertToSessions(ctx, sessionsList, withListingConvert(true)) if err != nil { return nil, errors.Wrap(ctx, err, op) } @@ -790,3 +798,15 @@ func fetchStates(ctx context.Context, r db.Reader, sessionId string, opt ...db.O } return states, nil } + +func fetchConnections(ctx context.Context, r db.Reader, sessionId string, opt ...db.Option) ([]*Connection, error) { + const op = "session.fetchConnections" + var connections []*Connection + if err := r.SearchWhere(ctx, &connections, "session_id = ?", []interface{}{sessionId}, opt...); err != nil { + return nil, errors.Wrap(ctx, err, op) + } + if len(connections) == 0 { + return nil, nil + } + return connections, nil +} diff --git a/internal/session/repository_session_test.go b/internal/session/repository_session_test.go index 1bc59ee9d..d6bea02a6 100644 --- a/internal/session/repository_session_test.go +++ b/internal/session/repository_session_test.go @@ -48,11 +48,12 @@ func TestRepository_ListSession(t *testing.T) { opt []Option } tests := []struct { - name string - createCnt int - args args - wantCnt int - wantErr bool + name string + createCnt int + args args + wantCnt int + wantErr bool + withConnections int }{ { name: "no-limit", @@ -97,6 +98,14 @@ func TestRepository_ListSession(t *testing.T) { wantCnt: 0, wantErr: false, }, + { + name: "multiple-connections", + createCnt: repo.defaultLimit + 1, + args: args{}, + wantCnt: repo.defaultLimit, + wantErr: false, + withConnections: 3, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -107,6 +116,9 @@ func TestRepository_ListSession(t *testing.T) { s := TestSession(t, conn, wrapper, composedOf) _ = TestState(t, conn, s.PublicId, StatusActive) testSessions = append(testSessions, s) + for i := 0; i < tt.withConnections; i++ { + _ = TestConnection(t, conn, s.PublicId, "127.0.0.1", 22, "127.0.0.2", 23) + } } assert.Equal(tt.createCnt, len(testSessions)) got, err := repo.ListSessions(context.Background(), tt.args.opt...) @@ -116,6 +128,15 @@ func TestRepository_ListSession(t *testing.T) { } require.NoError(err) assert.Equal(tt.wantCnt, len(got)) + for i := 0; i < len(got); i++ { + assert.Equal(tt.withConnections, len(got[i].Connections)) + for _, c := range got[i].Connections { + assert.Equal("127.0.0.1", c.ClientTcpAddress) + assert.Equal(uint32(22), c.ClientTcpPort) + assert.Equal("127.0.0.2", c.EndpointTcpAddress) + assert.Equal(uint32(23), c.EndpointTcpPort) + } + } if tt.wantCnt > 0 { assert.Equal(StatusActive, got[0].States[0].Status) assert.Equal(StatusPending, got[0].States[1].Status) diff --git a/internal/session/session.go b/internal/session/session.go index 514028cbf..cf0864b6f 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -112,6 +112,9 @@ type Session struct { // DynamicCredentials for the session. DynamicCredentials []*DynamicCredential `gorm:"-"` + // Connections for the session are for read only and are ignored during write operations + Connections []*Connection `gorm:"-"` + tableName string `gorm:"-"` } @@ -401,7 +404,7 @@ func (s *Session) decrypt(ctx context.Context, cipher wrapping.Wrapper) error { return nil } -type sessionView struct { +type sessionListView struct { // Session fields PublicId string `json:"public_id,omitempty" gorm:"primary_key"` UserId string `json:"user_id,omitempty" gorm:"default:null"` @@ -429,9 +432,19 @@ type sessionView struct { PreviousEndTime *timestamp.Timestamp `json:"previous_end_time,omitempty" gorm:"default:current_timestamp"` StartTime *timestamp.Timestamp `json:"start_time,omitempty" gorm:"default:current_timestamp;primary_key"` EndTime *timestamp.Timestamp `json:"end_time,omitempty" gorm:"default:current_timestamp"` + + // Connection fields + ConnectionId string `json:"connection_id,omitempty" gorm:"default:null"` + ClientTcpAddress string `json:"client_tcp_address,omitempty" gorm:"default:null"` + ClientTcpPort uint32 `json:"client_tcp_port,omitempty" gorm:"default:null"` + EndpointTcpAddress string `json:"endpoint_tcp_address,omitempty" gorm:"default:null"` + EndpointTcpPort uint32 `json:"endpoint_tcp_port,omitempty" gorm:"default:null"` + BytesUp uint64 `json:"bytes_up,omitempty" gorm:"default:null"` + BytesDown uint64 `json:"bytes_down,omitempty" gorm:"default:null"` + ClosedReason string `json:"closed_reason,omitempty" gorm:"default:null"` } // TableName returns the tablename to override the default gorm table name -func (s *sessionView) TableName() string { - return "session_with_state" +func (s *sessionListView) TableName() string { + return "session_list" }