This repository has been archived by the owner on Apr 18, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 9
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
bb6c08d
commit dc0bcb4
Showing
12 changed files
with
474 additions
and
37 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,135 @@ | ||
| package carserver | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/json" | ||
| "fmt" | ||
| "sync" | ||
|
|
||
| "github.com/ipfs/go-datastore/namespace" | ||
|
|
||
| datastore "github.com/ipfs/go-datastore" | ||
|
|
||
| "github.com/filecoin-project/saturn-l2/station" | ||
| ) | ||
|
|
||
| var Version = "v0.0.0" | ||
| var contentReqKey = datastore.NewKey("/content-req") | ||
| var storeNameSpace = "station" | ||
|
|
||
| type StationAPIImpl struct { | ||
| ss station.StorageStatsFetcher | ||
|
|
||
| mu sync.RWMutex | ||
| ds datastore.Batching | ||
| } | ||
|
|
||
| func NewStationAPIImpl(ds datastore.Batching, ss station.StorageStatsFetcher) *StationAPIImpl { | ||
| nds := namespace.Wrap(ds, datastore.NewKey(storeNameSpace)) | ||
| return &StationAPIImpl{ | ||
| ss: ss, | ||
| ds: nds, | ||
| } | ||
| } | ||
|
|
||
| func (s *StationAPIImpl) SetStorageStatsFetcher(ss station.StorageStatsFetcher) { | ||
| s.ss = ss | ||
| } | ||
|
|
||
| func (s *StationAPIImpl) RecordRetrievalServed(ctx context.Context, bytesServed, nErrors uint64) error { | ||
| s.mu.Lock() | ||
| defer s.mu.Unlock() | ||
|
|
||
| return s.createOrUpdateReqStatsUnlocked(ctx, func(r *station.ReqStats) { | ||
| r.Upload = bytesServed | ||
| r.ContentRequests = 1 | ||
| r.ContentReqErrors = nErrors | ||
| }, func(r *station.ReqStats) { | ||
| r.Upload += bytesServed | ||
| r.ContentRequests += 1 | ||
| r.ContentReqErrors += nErrors | ||
| }) | ||
| } | ||
|
|
||
| func (s *StationAPIImpl) RecordDataDownloaded(ctx context.Context, bytesDownloaded uint64) error { | ||
| s.mu.Lock() | ||
| defer s.mu.Unlock() | ||
|
|
||
| return s.createOrUpdateReqStatsUnlocked(ctx, func(r *station.ReqStats) { | ||
| r.Download = bytesDownloaded | ||
| }, func(r *station.ReqStats) { | ||
| r.Download += bytesDownloaded | ||
| }) | ||
| } | ||
|
|
||
| func (s *StationAPIImpl) createOrUpdateReqStatsUnlocked(ctx context.Context, createFn func(s *station.ReqStats), | ||
| updateFn func(s *station.ReqStats)) error { | ||
|
|
||
| bz, err := s.ds.Get(ctx, contentReqKey) | ||
| if err != nil && err != datastore.ErrNotFound { | ||
| return fmt.Errorf("failed to get retrieval stats from datastore: %w", err) | ||
| } | ||
| if err == datastore.ErrNotFound { | ||
| stats := station.ReqStats{} | ||
| createFn(&stats) | ||
| bz, err := json.Marshal(stats) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to marshal retrieval stats to json: %w", err) | ||
| } | ||
|
|
||
| return s.ds.Put(ctx, contentReqKey, bz) | ||
| } | ||
| var stats station.ReqStats | ||
| if err := json.Unmarshal(bz, &stats); err != nil { | ||
| return fmt.Errorf("failed to unmarshal existing retrieval stats: %w", err) | ||
| } | ||
|
|
||
| updateFn(&stats) | ||
|
|
||
| bz, err = json.Marshal(stats) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to marshal retrieval stats to json: %w", err) | ||
| } | ||
| return s.ds.Put(ctx, contentReqKey, bz) | ||
| } | ||
|
|
||
| func (s *StationAPIImpl) AllStats(ctx context.Context) (station.StationStats, error) { | ||
| s.mu.RLock() | ||
| defer s.mu.RUnlock() | ||
|
|
||
| // storage stats | ||
| storage, err := s.ss.Stat() | ||
| if err != nil { | ||
| return station.StationStats{}, fmt.Errorf("failed to fetch storage stats: %w", err) | ||
| } | ||
|
|
||
| // info | ||
| info := station.RPInfo{ | ||
| Version: Version, | ||
| } | ||
|
|
||
| // fetch retrieval stats | ||
| bz, err := s.ds.Get(ctx, contentReqKey) | ||
| if err != nil && err != datastore.ErrNotFound { | ||
| return station.StationStats{}, fmt.Errorf("failed to fetch retrieval stats: %w", err) | ||
| } | ||
| if err == datastore.ErrNotFound { | ||
| return station.StationStats{ | ||
| RPInfo: info, | ||
| StorageStats: storage, | ||
| }, nil | ||
| } | ||
|
|
||
| var rs station.ReqStats | ||
| if err := json.Unmarshal(bz, &rs); err != nil { | ||
| return station.StationStats{}, fmt.Errorf("failed to unmarshal retrieval stats from json: %w", err) | ||
| } | ||
|
|
||
| return station.StationStats{ | ||
| RPInfo: info, | ||
| StorageStats: storage, | ||
| ReqStats: rs, | ||
| }, nil | ||
| } | ||
|
|
||
| var _ station.StationAPI = &StationAPIImpl{} |
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,85 @@ | ||
| package carserver | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
|
|
||
| "github.com/filecoin-project/saturn-l2/station" | ||
| datastore "github.com/ipfs/go-datastore" | ||
| dss "github.com/ipfs/go-datastore/sync" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestStationAPIImpl(t *testing.T) { | ||
| ctx := context.Background() | ||
| ds := dss.MutexWrap(datastore.NewMapDatastore()) | ||
| sapi := NewStationAPIImpl(ds, &mockStorageStatsFetcher{ | ||
| out: 790, | ||
| }) | ||
|
|
||
| as, err := sapi.AllStats(ctx) | ||
| require.NoError(t, err) | ||
| require.Equal(t, station.StationStats{RPInfo: station.RPInfo{Version: Version}, | ||
| StorageStats: station.StorageStats{ | ||
| Bytes: 790, | ||
| }}, as) | ||
|
|
||
| require.NoError(t, sapi.RecordDataDownloaded(ctx, 100)) | ||
| as, err = sapi.AllStats(ctx) | ||
| require.NoError(t, err) | ||
| require.Equal(t, station.StationStats{RPInfo: station.RPInfo{Version: Version}, | ||
| StorageStats: station.StorageStats{ | ||
| Bytes: 790, | ||
| }, | ||
| ReqStats: station.ReqStats{ | ||
| Download: 100, | ||
| }}, as) | ||
|
|
||
| require.NoError(t, sapi.RecordDataDownloaded(ctx, 200)) | ||
| as, err = sapi.AllStats(ctx) | ||
| require.NoError(t, err) | ||
| require.Equal(t, station.StationStats{RPInfo: station.RPInfo{Version: Version}, | ||
| StorageStats: station.StorageStats{ | ||
| Bytes: 790, | ||
| }, | ||
| ReqStats: station.ReqStats{ | ||
| Download: 300, | ||
| }}, as) | ||
|
|
||
| require.NoError(t, sapi.RecordRetrievalServed(ctx, 100, 0)) | ||
| as, err = sapi.AllStats(ctx) | ||
| require.NoError(t, err) | ||
| require.Equal(t, station.StationStats{RPInfo: station.RPInfo{Version: Version}, | ||
| StorageStats: station.StorageStats{ | ||
| Bytes: 790, | ||
| }, | ||
| ReqStats: station.ReqStats{ | ||
| Upload: 100, | ||
| ContentRequests: 1, | ||
| Download: 300, | ||
| }}, as) | ||
|
|
||
| require.NoError(t, sapi.RecordRetrievalServed(ctx, 500, 2)) | ||
| as, err = sapi.AllStats(ctx) | ||
| require.NoError(t, err) | ||
| require.Equal(t, station.StationStats{RPInfo: station.RPInfo{Version: Version}, | ||
| StorageStats: station.StorageStats{ | ||
| Bytes: 790, | ||
| }, | ||
| ReqStats: station.ReqStats{ | ||
| Upload: 600, | ||
| ContentRequests: 2, | ||
| ContentReqErrors: 2, | ||
| Download: 300, | ||
| }}, as) | ||
| } | ||
|
|
||
| type mockStorageStatsFetcher struct { | ||
| out uint64 | ||
| } | ||
|
|
||
| func (ms *mockStorageStatsFetcher) Stat() (station.StorageStats, error) { | ||
| return station.StorageStats{ | ||
| Bytes: ms.out, | ||
| }, nil | ||
| } |
Oops, something went wrong.