diff --git a/internal/bsr/bsr.go b/internal/bsr/bsr.go index 613f24124..a5eee4ceb 100644 --- a/internal/bsr/bsr.go +++ b/internal/bsr/bsr.go @@ -648,7 +648,12 @@ func (c *Connection) NewMessagesWriter(ctx context.Context, dir Direction) (io.W if err != nil { return nil, err } - m, err := c.container.create(ctx, messagesName) + // 65*storage.LogicalBlockSize is equivalent to 260KiB + m, err := c.container.create(ctx, messagesName, + storage.WithCreateFile(), + storage.WithFileAccessMode(storage.ReadWrite), + storage.WithBuffer(65*storage.LogicalBlockSize), + ) if err != nil { return nil, err } @@ -670,7 +675,12 @@ func (c *Connection) NewRequestsWriter(ctx context.Context, dir Direction) (stor if err != nil { return nil, err } - m, err := c.container.create(ctx, requestName) + // 65*storage.LogicalBlockSize is equivalent to 260KiB + m, err := c.container.create(ctx, requestName, + storage.WithCreateFile(), + storage.WithFileAccessMode(storage.ReadWrite), + storage.WithBuffer(65*storage.LogicalBlockSize), + ) if err != nil { return nil, err } diff --git a/internal/bsr/container.go b/internal/bsr/container.go index 017551e17..53d3ad13d 100644 --- a/internal/bsr/container.go +++ b/internal/bsr/container.go @@ -28,11 +28,117 @@ const ( checksumFileName = "SHA256SUM" sigFileName = "SHA256SUM.sig" journalFileName = ".journal" + + // sigFileBufferSize represents the number of free bytes required in the CHECKSUM.sig file to close safely + sigFileBufferSize = 89 + // checksumFileBufferSize represents the number of free bytes required in the CHECKSUM file to safely + // write a single line: "{hash_value} {file_name}\n" + checksumFileBufferSize = 320 + // sessionRecordingSummaryBufferSize represents the number of free bytes required in the session-recording-summary.json + // to close safely. This value is determined by the BaseSessionSummary definition. + sessionRecordingSummaryBufferSize = 183 + // connectionRecordingSummaryBufferSize represents the number of free bytes required in the connection-recording-summary.json + // to close safely. This value is determined by the BaseConnectionSummary definition. + connectionRecordingSummaryBufferSize = 252 + // connectionRecordingSummaryBufferSize represents the number of free bytes required in the channel-recording-summary.json + // to close safely. This value is determined by the ChannelSummary definition. + channelRecordingSummaryBufferSize = 518 + // sessionRecordingMetaBufferSize represents the number of free bytes required in the session-recording.meta to safely + // write a single line: "connection: {connection_recorder_id}" + sessionRecordingMetaBufferSize = 36 + // connectionRecordingMetaBufferSize represents the number of free bytes required in the connection-recording.meta to safely + // write a single line: "channel: {channel_recorder_id}" + connnectionRecordingMetaBufferSize = 31 + // channelRecordingMetaBufferSize represents the number of free bytes required in the channel-recording.meta + // to close safely. This value is determined by the ChannelRecordingMeta definition. + channelRecordingMetaBufferSize = 114 + // sessionJournalBufferSize represents the number of free bytes required in the session container .journal file to + // close safely. This value is determined by the number of files that exist in a session container. + sessionJournalBufferSize = 1060 + // connectionJournalBufferSize represents the number of free bytes required in the connection container .journal file to + // close safely. This value is determined by the number of files that exist in a connection container. + connectionJournalBufferSize = 708 + // connectionJournalBufferSize represents the number of free bytes required in the channel container .journal file to + // close safely. This value is determined by the number of files that exist in a channel container. + channelJournalBufferSize = 928 ) // ContainerType defines the type of container. type ContainerType string +// ChecksumBufferSize returns the buffer size needed +// for the rolling buffer of a checksum file +func (c ContainerType) ChecksumBufferSize() (uint64, error) { + switch c { + case SessionContainer: + fallthrough + case ConnectionContainer: + fallthrough + case ChannelContainer: + return checksumFileBufferSize, nil + } + return 0, fmt.Errorf("unknown container type: %s", c) +} + +// ChecksumSignatureBufferSize returns the buffer size +// needed for the rolling buffer of a checksum file +func (c ContainerType) ChecksumSignatureBufferSize() (uint64, error) { + switch c { + case SessionContainer: + fallthrough + case ConnectionContainer: + fallthrough + case ChannelContainer: + return sigFileBufferSize, nil + } + return 0, fmt.Errorf("unknown container type: %s", c) +} + +// RecordingSummaryBufferSize returns the buffer size needed +// for the rolling buffer of a recording summary file based +// on the container type +func (c ContainerType) RecordingSummaryBufferSize() (uint64, error) { + switch c { + case SessionContainer: + return sessionRecordingSummaryBufferSize, nil + case ConnectionContainer: + return connectionRecordingSummaryBufferSize, nil + case ChannelContainer: + return channelRecordingSummaryBufferSize, nil + } + return 0, fmt.Errorf("unknown container type: %s", c) +} + +// RecordingMetaBufferSize returns the buffer size needed +// for the rolling buffer of a recording meta file based +// on the container type +func (c ContainerType) RecordingMetaBufferSize() (uint64, error) { + switch c { + case SessionContainer: + return sessionRecordingMetaBufferSize, nil + case ConnectionContainer: + return connnectionRecordingMetaBufferSize, nil + case ChannelContainer: + return channelRecordingMetaBufferSize, nil + } + return 0, fmt.Errorf("unknown container type: %s", c) +} + +// JournalBufferSize returns the buffer size needed +// for the rolling buffer for a journal file based +// on the container type +func (c ContainerType) JournalBufferSize() (uint64, error) { + switch c { + case SessionContainer: + return sessionJournalBufferSize, nil + case ConnectionContainer: + return connectionJournalBufferSize, nil + case ChannelContainer: + return channelJournalBufferSize, nil + } + return 0, fmt.Errorf("unknown container type: %s", c) +} + // FileChecksumValidation is a validation report on a file's checksum value type FileChecksumValidation struct { Filename string @@ -89,10 +195,16 @@ type container struct { // newContainer creates a container for the given type backed by the provide storage.Container. func newContainer(ctx context.Context, t ContainerType, c storage.Container, keys *kms.Keys) (*container, error) { + journalBufferSize, err := t.JournalBufferSize() + if err != nil { + return nil, err + } j, err := c.OpenFile(ctx, journalFileName, storage.WithCreateFile(), - storage.WithFileAccessMode(storage.WriteOnly), - storage.WithCloseSyncMode(storage.NoSync)) + storage.WithFileAccessMode(storage.ReadWrite), + storage.WithCloseSyncMode(storage.NoSync), + storage.WithBuffer(journalBufferSize), + ) if err != nil { return nil, err } @@ -106,12 +218,28 @@ func newContainer(ctx context.Context, t ContainerType, c storage.Container, key keys: keys, } - cc.sigs, err = cc.create(ctx, sigFileName) + sigFileBufferSize, err := t.ChecksumSignatureBufferSize() + if err != nil { + return nil, err + } + cc.sigs, err = cc.create(ctx, sigFileName, + storage.WithCreateFile(), + storage.WithFileAccessMode(storage.ReadWrite), + storage.WithBuffer(sigFileBufferSize), + ) if err != nil { return nil, err } - cs, err := cc.create(ctx, checksumFileName) + checksumFileBufferSize, err := t.ChecksumBufferSize() + if err != nil { + return nil, err + } + cs, err := cc.create(ctx, checksumFileName, + storage.WithCreateFile(), + storage.WithFileAccessMode(storage.ReadWrite), + storage.WithBuffer(checksumFileBufferSize), + ) if err != nil { return nil, err } @@ -120,8 +248,16 @@ func newContainer(ctx context.Context, t ContainerType, c storage.Container, key return nil, err } + metaBufferSize, err := t.RecordingMetaBufferSize() + if err != nil { + return nil, err + } cc.metaName = fmt.Sprintf(metaFileNameTemplate, t) - meta, err := cc.create(ctx, cc.metaName) + meta, err := cc.create(ctx, cc.metaName, + storage.WithCreateFile(), + storage.WithFileAccessMode(storage.ReadWrite), + storage.WithBuffer(metaBufferSize), + ) if err != nil { return nil, err } @@ -130,8 +266,16 @@ func newContainer(ctx context.Context, t ContainerType, c storage.Container, key return nil, err } + summaryBufferSize, err := t.RecordingSummaryBufferSize() + if err != nil { + return nil, err + } cc.sumName = fmt.Sprintf(summaryFileNameTemplate, t) - sum, err := cc.create(ctx, cc.sumName) + sum, err := cc.create(ctx, cc.sumName, + storage.WithCreateFile(), + storage.WithFileAccessMode(storage.ReadWrite), + storage.WithBuffer(summaryBufferSize), + ) if err != nil { return nil, err } diff --git a/internal/bsr/ssh/buffer_size_test.go b/internal/bsr/ssh/buffer_size_test.go new file mode 100644 index 000000000..454560cb5 --- /dev/null +++ b/internal/bsr/ssh/buffer_size_test.go @@ -0,0 +1,220 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package ssh + +import ( + "context" + "fmt" + "io" + "math" + "testing" + "time" + + "github.com/hashicorp/boundary/globals" + "github.com/hashicorp/boundary/internal/bsr" + "github.com/hashicorp/boundary/internal/bsr/internal/fstest" + "github.com/hashicorp/boundary/internal/bsr/kms" + "github.com/hashicorp/boundary/internal/db" + "github.com/hashicorp/boundary/internal/storage" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Note, we cannot validate the allocation of the rolling buffer +// because these tests are utilizing MemFiles and not SyncingFiles. +// MemFile does not implement a rolling buffer. +// This test validates that the minimum buffer size for each file +// type is accurate. +func TestValidateBufferSize(t *testing.T) { + protocol := bsr.Protocol("TEST") + bsr.RegisterSummaryAllocFunc(protocol, bsr.SessionContainer, func(ctx context.Context) bsr.Summary { + return &bsr.BaseSessionSummary{Id: "sr_123456789", ConnectionCount: 1} + }) + bsr.RegisterSummaryAllocFunc(protocol, bsr.ConnectionContainer, func(ctx context.Context) bsr.Summary { + return &bsr.BaseConnectionSummary{Id: "cr_123456789", ChannelCount: 1} + }) + bsr.RegisterSummaryAllocFunc(protocol, bsr.ChannelContainer, func(ctx context.Context) bsr.Summary { + return &bsr.BaseChannelSummary{Id: "chr_123456789", ConnectionRecordingId: "cr_123456789"} + }) + + ctx := context.Background() + ts := time.Date(1995, time.March, 3, 12, 12, 12, 999999999, time.UTC) + + sessionId, err := db.NewPublicId(ctx, globals.SessionRecordingPrefix) + require.NoError(t, err) + + connectionId, err := db.NewPublicId(ctx, globals.ConnectionRecordingPrefix) + require.NoError(t, err) + + channelId, err := db.NewPublicId(ctx, globals.ChannelRecordingPrefix) + require.NoError(t, err) + + containerName := fmt.Sprintf("%s.bsr", sessionId) + connectionName := fmt.Sprintf("%s.connection", connectionId) + channelName := fmt.Sprintf("%s.channel", channelId) + + var writer storage.Writer + fs := &fstest.MemFS{} + keys, err := kms.CreateKeys(ctx, kms.TestWrapper(t), sessionId) + require.NoError(t, err) + + // setup session container + session, err := bsr.NewSession(ctx, bsr.TestSessionRecordingMeta(sessionId, protocol), bsr.TestSessionMeta(sessionId), fs, keys, bsr.WithSupportsMultiplex(true)) + require.NoError(t, err) + require.NoError(t, session.EncodeSummary(ctx, &bsr.BaseSessionSummary{ + Id: sessionId, + ConnectionCount: math.MaxUint64, + StartTime: ts, + EndTime: ts, + Errors: bsr.SummaryError{ + Message: "", + }, + })) + + // setup connection container + connection, err := session.NewConnection(ctx, &bsr.ConnectionRecordingMeta{ + Id: connectionId, + }) + require.NoError(t, err) + require.NoError(t, connection.EncodeSummary(ctx, &bsr.BaseConnectionSummary{ + Id: connectionId, + ChannelCount: math.MaxUint64, + BytesUp: math.MaxUint64, + BytesDown: math.MaxUint64, + StartTime: ts, + EndTime: ts, + Errors: bsr.SummaryError{ + Message: "", + }, + })) + // create connection inbound request writer + writer, err = connection.NewRequestsWriter(ctx, bsr.Inbound) + require.NoError(t, err) + closer, ok := writer.(io.Closer) + require.True(t, ok) + require.NoError(t, closer.Close()) + // create connection outbound request writer + writer, err = connection.NewRequestsWriter(ctx, bsr.Outbound) + require.NoError(t, err) + closer, ok = writer.(io.Closer) + require.True(t, ok) + require.NoError(t, closer.Close()) + + // create channel container + channel, err := connection.NewChannel(ctx, &bsr.ChannelRecordingMeta{ + Id: channelId, + Type: "session", + }) + require.NoError(t, err) + require.NoError(t, channel.EncodeSummary(ctx, &ChannelSummary{ + ChannelSummary: &bsr.BaseChannelSummary{ + Id: channelId, + ConnectionRecordingId: connectionId, + BytesUp: math.MaxUint64, + BytesDown: math.MaxUint64, + StartTime: ts, + EndTime: ts, + ChannelType: "session", + }, + SessionProgram: NotApplicable, + SubsystemName: "", + ExecProgram: ExecApplicationProgramNotApplicable, + FileTransferDirection: FileTransferNotApplicable, + OpenFailure: &OpenChannelError{ + Reason: math.MaxUint32, + Message: "", + }, + })) + // create channel inbound request writer + writer, err = channel.NewRequestsWriter(ctx, bsr.Inbound) + require.NoError(t, err) + closer, ok = writer.(io.Closer) + require.True(t, ok) + require.NoError(t, closer.Close()) + // create channel outbound request writer + writer, err = channel.NewRequestsWriter(ctx, bsr.Outbound) + require.NoError(t, err) + closer, ok = writer.(io.Closer) + require.True(t, ok) + require.NoError(t, closer.Close()) + // create channel inbound message writer + writer, err = channel.NewMessagesWriter(ctx, bsr.Inbound) + require.NoError(t, err) + closer, ok = writer.(io.Closer) + require.True(t, ok) + require.NoError(t, closer.Close()) + // create channel outbound message writer + writer, err = channel.NewMessagesWriter(ctx, bsr.Outbound) + require.NoError(t, err) + closer, ok = writer.(io.Closer) + require.True(t, ok) + require.NoError(t, closer.Close()) + + // close containers + require.NoError(t, channel.Close(ctx)) + require.NoError(t, connection.Close(ctx)) + require.NoError(t, session.Close(ctx)) + + // validate CHECKSUM.sig buffer size + // this should always be 88 bytes for + // all container types + actualChecksumSignatureSize := fs.Containers[containerName].Files["SHA256SUM.sig"].Buf.Len() + expectedChecksumSignatureBufferSize, err := bsr.SessionContainer.ChecksumSignatureBufferSize() + require.NoError(t, err) + assert.EqualValues(t, expectedChecksumSignatureBufferSize, actualChecksumSignatureSize) + + // validate CHECKSUM buffer size + // this should always be 320 bytes + // for all container types. This + // ensures a single line can be written: + // hash + empty space + file name + expectedChecksumBufferSize, err := bsr.SessionContainer.ChecksumBufferSize() + require.NoError(t, err) + assert.EqualValues(t, expectedChecksumBufferSize, len("dc8ce2c42553ce510197c99efe21d89d6229feb4b49170511f49965f2e3cf1a3 ")+255) + + cases := []struct { + name string + containerType bsr.ContainerType + actualJournalSize int + actualRecordingMetaSize int + actualRecordingSummarySize int + }{ + { + name: "session", + containerType: bsr.SessionContainer, + actualJournalSize: fs.Containers[containerName].Files[".journal"].Buf.Len(), + actualRecordingMetaSize: len(fmt.Sprintf("connection: %s.connection", connectionId)), + actualRecordingSummarySize: fs.Containers[containerName].Files["session-recording-summary.json"].Buf.Len(), + }, + { + name: "connection", + containerType: bsr.ConnectionContainer, + actualJournalSize: fs.Containers[containerName].Sub[connectionName].Files[".journal"].Buf.Len(), + actualRecordingMetaSize: len(fmt.Sprintf("channel: %s.channel", channelId)), + actualRecordingSummarySize: fs.Containers[containerName].Sub[connectionName].Files["connection-recording-summary.json"].Buf.Len(), + }, + { + name: "channel", + containerType: bsr.ChannelContainer, + actualJournalSize: fs.Containers[containerName].Sub[connectionName].Sub[channelName].Files[".journal"].Buf.Len(), + actualRecordingMetaSize: fs.Containers[containerName].Sub[connectionName].Sub[channelName].Files["channel-recording.meta"].Buf.Len(), + actualRecordingSummarySize: fs.Containers[containerName].Sub[connectionName].Sub[channelName].Files["channel-recording-summary.json"].Buf.Len(), + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + expectedJournalBufferSize, err := tc.containerType.JournalBufferSize() + require.NoError(t, err) + assert.EqualValues(t, expectedJournalBufferSize, tc.actualJournalSize) + + expectedRecordingMetaBufferSize, err := tc.containerType.RecordingMetaBufferSize() + require.NoError(t, err) + assert.EqualValues(t, expectedRecordingMetaBufferSize, tc.actualRecordingMetaSize) + + expectedRecordingSummaryBufferSize, err := tc.containerType.RecordingSummaryBufferSize() + require.NoError(t, err) + assert.EqualValues(t, expectedRecordingSummaryBufferSize, tc.actualRecordingSummarySize) + }) + } +}