feat(bsr): utilize rolling buffer
This commit is contained in:
Damian Debkowski authored and 
Elim Tsiagbey committed
1 parent
5a6a6a5198
commit
4f42dbf59f
3 files changed
+382
-8
No files matched your search
+12
-2
@@ -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
|
||||
}
|
||||
|
||||
+150
-6
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user