-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #1 from risingwavelabs/feat/stats
feat: expose stats
- Loading branch information
Showing
7 changed files
with
206 additions
and
8 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,48 @@ | ||
| // Copyright 2024 RisingWave Labs | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package filechannel | ||
|
|
||
| // ReceiverStats is the interface for getting the stats of a receiver. | ||
| type ReceiverStats interface { | ||
| // ReadOffset returns the offset of the last read message. | ||
| // Note that the offset is local to the receiver. It will only change | ||
| // when the receiver reads a message. | ||
| // | ||
| // The initial offset is math.MaxUint64 to indicate that no message has | ||
| // been read. | ||
| ReadOffset() uint64 | ||
| } | ||
|
|
||
| // SenderStats is the interface for getting the stats of a sender. | ||
| type SenderStats interface { | ||
| // WriteOffset returns the offset of the last written message. | ||
| // Note that the offset is not the byte offset in the file. It's the | ||
| // offset of the message in the channel. The offset will change no matter | ||
| // a message is written by the sender, or the other senders of the same | ||
| // channel. | ||
| WriteOffset() uint64 | ||
| } | ||
|
|
||
| // Stats is the interface for getting the stats of a file channel. | ||
| type Stats interface { | ||
| // DiskUsage returns the disk usage of the file channel. | ||
| // Note that calling DiskUsage() is an expensive operation. | ||
| DiskUsage() (uint64, error) | ||
|
|
||
| // FlushOffset returns the offset of the last flushed message. | ||
| // Messages with offset less than the flush offset are guaranteed to be | ||
| // seen by the readers. | ||
| FlushOffset() uint64 | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| // Copyright 2024 RisingWave Labs | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package filechannel | ||
|
|
||
| import ( | ||
| "context" | ||
| "math" | ||
| "os" | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| func TestFileChannel_Stats(t *testing.T) { | ||
| tmpDir := mkdirTemp(t) | ||
| defer os.RemoveAll(tmpDir) | ||
|
|
||
| fch, err := OpenFileChannel(tmpDir) | ||
| if !assert.NoError(t, err) { | ||
| t.FailNow() | ||
| } | ||
| defer fch.Close() | ||
|
|
||
| // Assert disk usage == 64 (header size). | ||
| usage, err := fch.DiskUsage() | ||
| if !assert.NoError(t, err) { | ||
| t.FailNow() | ||
| } | ||
| assert.Equal(t, uint64(64), usage) | ||
|
|
||
| msg := []byte("Hello world!") | ||
|
|
||
| tx := fch.Tx() | ||
| defer tx.Close() | ||
|
|
||
| // Assert sender offset == 0. | ||
| assert.Equal(t, uint64(0), tx.WriteOffset()) | ||
|
|
||
| err = tx.Send(context.Background(), msg) | ||
| if !assert.NoError(t, err) { | ||
| t.FailNow() | ||
| } | ||
|
|
||
| // Assert sender offset != 0. | ||
| assert.NotEqual(t, uint64(0), tx.WriteOffset()) | ||
|
|
||
| rx := fch.Rx() | ||
| defer rx.Close() | ||
|
|
||
| // Assert reader offset == math.MaxUint64. | ||
| assert.Equal(t, uint64(math.MaxUint64), rx.ReadOffset()) | ||
|
|
||
| p, err := rx.Recv(context.Background()) | ||
| if !assert.NoError(t, err) { | ||
| t.FailNow() | ||
| } | ||
| if !assert.Equal(t, msg, p) { | ||
| t.FailNow() | ||
| } | ||
|
|
||
| // Assert reader offset != 0. | ||
| assert.NotEqual(t, uint64(0), rx.ReadOffset()) | ||
|
|
||
| // Assert reader offset == sender offset. | ||
| assert.Equal(t, tx.WriteOffset(), rx.ReadOffset()) | ||
|
|
||
| // Recv happened means the file channel has flushed. | ||
| // Now we can examine the disk usage and flush offset. | ||
|
|
||
| // Assert flush offset != 0. | ||
| assert.NotEqual(t, uint64(0), fch.FlushOffset()) | ||
|
|
||
| // Assert disk usage > 64. | ||
| usage, err = fch.DiskUsage() | ||
| if !assert.NoError(t, err) { | ||
| t.FailNow() | ||
| } | ||
| assert.NotEqual(t, uint64(64), usage) | ||
| } |