vincent/aptly
1
0
Fork 0
mirror of https://github.com/aptly-dev/aptly.git synced 2026-07-15 11:57:59 +00:00
Code Issues Projects Releases Wiki Activity

Fix S3 concurrent map writes causing pod crashes

Browse Source 
PROBLEM:
- Pod crashes with "fatal error: concurrent map writes" during S3 publications
- Root cause: pathCache map in PublishedStorage accessed without synchronization
- Occurs during concurrent LinkFromPool operations in S3 publishing

SOLUTION:
- Add sync.RWMutex to PublishedStorage struct for thread-safe map access
- Implement double-check locking pattern for cache initialization
- Protect all map operations (read/write/delete) with appropriate locks

CHANGES:
- s3/public.go: Add pathCacheMutex field and protect all map operations
  * Cache initialization with double-check locking in LinkFromPool
  * Read operations protected with RLock/RUnlock
  * Write operations protected with Lock/Unlock
  * Delete operations in Remove() and RemoveDirs() protected

IMPACT:
- Eliminates concurrent map writes panic
- Prevents pod crashes during S3 publications
- Maintains performance with minimal synchronization overhead
- Uses read-write locks allowing concurrent reads while serializing writes
This commit is contained in:
Nick Bozhenko
2025-07-15 22:46:37 -04:00
parent 308fb80a6e
commit 1693863499
102 changed files with 867 additions and 28632 deletions
Download Patch File Download Diff File Expand all files Collapse all files
database/etcddb/batch.go
+1 -5
View File
@@ -1,8 +1,6 @@
package etcddb
import (
"context"
"github.com/aptly-dev/aptly/database"
clientv3 "go.etcd.io/etcd/client/v3"
)
@@ -32,9 +30,7 @@ func (b *EtcDBatch) Write() (err error) {
batchSize := 128
for i := 0; i < len(b.ops); i += batchSize {
ctx, cancel := context.WithTimeout(context.Background(), DefaultTimeout)
defer cancel()
txn := kv.Txn(ctx)
txn := kv.Txn(Ctx)
end := i + batchSize
if end > len(b.ops) {
end = len(b.ops)
database/etcddb/database.go
+6 -52
View File
@@ -1,69 +1,23 @@
package etcddb
import (
"os"
"strconv"
"context"
"time"
"github.com/aptly-dev/aptly/database"
"github.com/rs/zerolog/log"
clientv3 "go.etcd.io/etcd/client/v3"
)
// Default timeout for etcd operations
var DefaultTimeout = 60 * time.Second
func init() {
// Allow timeout configuration via environment variable
if timeout := os.Getenv("APTLY_ETCD_TIMEOUT"); timeout != "" {
if d, err := time.ParseDuration(timeout); err == nil {
DefaultTimeout = d
log.Info().Dur("timeout", d).Msg("etcd: using custom timeout")
} else {
log.Warn().Str("value", timeout).Err(err).Msg("etcd: invalid timeout value, using default")
}
}
}
var Ctx = context.TODO()
func internalOpen(url string) (cli *clientv3.Client, err error) {
// Configure dial timeout
dialTimeout := 60 * time.Second
if dt := os.Getenv("APTLY_ETCD_DIAL_TIMEOUT"); dt != "" {
if d, err := time.ParseDuration(dt); err == nil {
dialTimeout = d
}
}
// Configure keep alive timeout
keepAliveTimeout := 7200 * time.Second
if ka := os.Getenv("APTLY_ETCD_KEEPALIVE"); ka != "" {
if d, err := time.ParseDuration(ka); err == nil {
keepAliveTimeout = d
}
}
// Configure message size
maxMsgSize := 50 * 1024 * 1024 // 50MiB default
if size := os.Getenv("APTLY_ETCD_MAX_MSG_SIZE"); size != "" {
if s, err := strconv.Atoi(size); err == nil && s > 0 {
maxMsgSize = s
}
}
cfg := clientv3.Config{
Endpoints: []string{url},
DialTimeout: dialTimeout,
MaxCallSendMsgSize: maxMsgSize,
MaxCallRecvMsgSize: maxMsgSize,
DialKeepAliveTimeout: keepAliveTimeout,
DialTimeout: 30 * time.Second,
MaxCallSendMsgSize: 2147483647, // (2048 * 1024 * 1024) - 1
MaxCallRecvMsgSize: 2147483647,
DialKeepAliveTimeout: 7200 * time.Second,
}
log.Info().
Str("endpoint", url).
Dur("dialTimeout", dialTimeout).
Dur("keepAlive", keepAliveTimeout).
Int("maxMsgSize", maxMsgSize).
Msg("etcd: opening connection")
cli, err = clientv3.New(cfg)
return
database/etcddb/storage.go
+10 -102
View File
@@ -1,14 +1,10 @@
package etcddb
import (
"context"
"fmt"
"strings"
"time"
"github.com/aptly-dev/aptly/database"
"github.com/google/uuid"
"github.com/rs/zerolog/log"
clientv3 "go.etcd.io/etcd/client/v3"
)
@@ -35,66 +31,11 @@ func (s *EtcDStorage) applyPrefix(key []byte) []byte {
return key
}
// getContext returns a context with timeout for etcd operations
func (s *EtcDStorage) getContext() (context.Context, context.CancelFunc) {
return context.WithTimeout(context.Background(), DefaultTimeout)
}
// isTemporary checks if error is temporary and can be retried
func isTemporary(err error) bool {
if err == nil {
return false
}
// Check for context deadline exceeded
if err == context.DeadlineExceeded {
return true
}
// Check for etcd specific temporary errors
switch err {
case clientv3.ErrNoAvailableEndpoints:
return true
default:
// Check if error string contains temporary indicators
errStr := err.Error()
return strings.Contains(errStr, "temporary") ||
strings.Contains(errStr, "timeout") ||
strings.Contains(errStr, "unavailable") ||
strings.Contains(errStr, "connection refused")
}
}
// Get key value from etcd
func (s *EtcDStorage) Get(key []byte) (value []byte, err error) {
realKey := s.applyPrefix(key)
var getResp *clientv3.GetResponse
maxRetries := 3
for i := 0; i < maxRetries; i++ {
ctx, cancel := s.getContext()
getResp, err = s.db.Get(ctx, string(realKey))
cancel()
if err == nil {
break
}
// Only retry on temporary errors and not on last attempt
if i < maxRetries-1 && isTemporary(err) {
backoff := time.Duration(i+1) * 100 * time.Millisecond
log.Warn().
Err(err).
Str("key", string(realKey)).
Int("attempt", i+1).
Dur("backoff", backoff).
Msg("etcd: get failed, retrying")
time.Sleep(backoff)
continue
}
log.Error().Err(err).Str("key", string(realKey)).Msg("etcd: get failed")
getResp, err := s.db.Get(Ctx, string(realKey))
if err != nil {
return
}
for _, kv := range getResp.Kvs {
@@ -111,13 +52,8 @@ func (s *EtcDStorage) Get(key []byte) (value []byte, err error) {
// Put saves key to etcd, if key has the same value in DB already, it is not saved
func (s *EtcDStorage) Put(key []byte, value []byte) (err error) {
realKey := s.applyPrefix(key)
ctx, cancel := s.getContext()
defer cancel()
_, err = s.db.Put(ctx, string(realKey), string(value))
_, err = s.db.Put(Ctx, string(realKey), string(value))
if err != nil {
log.Error().Err(err).Str("key", string(realKey)).Msg("etcd: put failed")
return
}
return
@@ -126,13 +62,8 @@ func (s *EtcDStorage) Put(key []byte, value []byte) (err error) {
// Delete removes key from etcd
func (s *EtcDStorage) Delete(key []byte) (err error) {
realKey := s.applyPrefix(key)
ctx, cancel := s.getContext()
defer cancel()
_, err = s.db.Delete(ctx, string(realKey))
_, err = s.db.Delete(Ctx, string(realKey))
if err != nil {
log.Error().Err(err).Str("key", string(realKey)).Msg("etcd: delete failed")
return
}
return
@@ -142,13 +73,8 @@ func (s *EtcDStorage) Delete(key []byte) (err error) {
func (s *EtcDStorage) KeysByPrefix(prefix []byte) [][]byte {
realPrefix := s.applyPrefix(prefix)
result := make([][]byte, 0, 20)
ctx, cancel := s.getContext()
defer cancel()
getResp, err := s.db.Get(ctx, string(realPrefix), clientv3.WithPrefix())
getResp, err := s.db.Get(Ctx, string(realPrefix), clientv3.WithPrefix())
if err != nil {
log.Error().Err(err).Str("prefix", string(realPrefix)).Msg("etcd: keys by prefix failed")
return nil
}
for _, ev := range getResp.Kvs {
@@ -164,13 +90,8 @@ func (s *EtcDStorage) KeysByPrefix(prefix []byte) [][]byte {
func (s *EtcDStorage) FetchByPrefix(prefix []byte) [][]byte {
realPrefix := s.applyPrefix(prefix)
result := make([][]byte, 0, 20)
ctx, cancel := s.getContext()
defer cancel()
getResp, err := s.db.Get(ctx, string(realPrefix), clientv3.WithPrefix())
getResp, err := s.db.Get(Ctx, string(realPrefix), clientv3.WithPrefix())
if err != nil {
log.Error().Err(err).Str("prefix", string(realPrefix)).Msg("etcd: fetch by prefix failed")
return nil
}
for _, kv := range getResp.Kvs {
@@ -185,13 +106,8 @@ func (s *EtcDStorage) FetchByPrefix(prefix []byte) [][]byte {
// HasPrefix checks whether it can find any key with given prefix and returns true if one exists
func (s *EtcDStorage) HasPrefix(prefix []byte) bool {
realPrefix := s.applyPrefix(prefix)
ctx, cancel := s.getContext()
defer cancel()
getResp, err := s.db.Get(ctx, string(realPrefix), clientv3.WithPrefix())
getResp, err := s.db.Get(Ctx, string(realPrefix), clientv3.WithPrefix())
if err != nil {
log.Error().Err(err).Str("prefix", string(realPrefix)).Msg("etcd: has prefix failed")
return false
}
return getResp.Count > 0
@@ -201,13 +117,8 @@ func (s *EtcDStorage) HasPrefix(prefix []byte) bool {
// StorageProcessor on key value pair
func (s *EtcDStorage) ProcessByPrefix(prefix []byte, proc database.StorageProcessor) error {
realPrefix := s.applyPrefix(prefix)
ctx, cancel := s.getContext()
defer cancel()
getResp, err := s.db.Get(ctx, string(realPrefix), clientv3.WithPrefix())
getResp, err := s.db.Get(Ctx, string(realPrefix), clientv3.WithPrefix())
if err != nil {
log.Error().Err(err).Str("prefix", string(realPrefix)).Msg("etcd: process by prefix failed")
return err
}
@@ -271,15 +182,12 @@ func (s *EtcDStorage) CompactDB() error {
// Drop removes only temporary DBs with etcd (i.e. remove all prefixed keys)
func (s *EtcDStorage) Drop() error {
if len(s.tmpPrefix) != 0 {
ctx, cancel := s.getContext()
defer cancel()
getResp, err := s.db.Get(ctx, s.tmpPrefix, clientv3.WithPrefix())
getResp, err := s.db.Get(Ctx, s.tmpPrefix, clientv3.WithPrefix())
if err != nil {
return nil
}
for _, kv := range getResp.Kvs {
_, err = s.db.Delete(ctx, string(kv.Key))
_, err = s.db.Delete(Ctx, string(kv.Key))
if err != nil {
return fmt.Errorf("cannot delete tempdb entry: %s", kv.Key)
}
database/etcddb/storage_test.go
-110
View File
@@ -1,110 +0,0 @@
package etcddb
import (
"context"
"os"
"testing"
"time"
. "gopkg.in/check.v1"
)
type StorageSuite struct{}
var _ = Suite(&StorageSuite{})
func Test(t *testing.T) { TestingT(t) }
func (s *StorageSuite) TestGetContext(c *C) {
storage := &EtcDStorage{}
// Test default timeout
ctx, cancel := storage.getContext()
defer cancel()
deadline, ok := ctx.Deadline()
c.Assert(ok, Equals, true)
// Should have a deadline set
remaining := time.Until(deadline)
c.Assert(remaining > 0, Equals, true)
c.Assert(remaining <= DefaultTimeout, Equals, true)
}
func (s *StorageSuite) TestDefaultTimeout(c *C) {
// Default should be 60 seconds
c.Assert(DefaultTimeout, Equals, 60*time.Second)
}
func (s *StorageSuite) TestEnvironmentVariables(c *C) {
// Save original values
originalTimeout := os.Getenv("APTLY_ETCD_TIMEOUT")
originalDialTimeout := os.Getenv("APTLY_ETCD_DIAL_TIMEOUT")
originalKeepAlive := os.Getenv("APTLY_ETCD_KEEPALIVE")
originalMaxMsg := os.Getenv("APTLY_ETCD_MAX_MSG_SIZE")
defer func() {
// Restore original values
os.Setenv("APTLY_ETCD_TIMEOUT", originalTimeout)
os.Setenv("APTLY_ETCD_DIAL_TIMEOUT", originalDialTimeout)
os.Setenv("APTLY_ETCD_KEEPALIVE", originalKeepAlive)
os.Setenv("APTLY_ETCD_MAX_MSG_SIZE", originalMaxMsg)
}()
// Test valid timeout
os.Setenv("APTLY_ETCD_TIMEOUT", "30s")
// Would need to reinitialize to test, but we can't easily do that
// This test mainly ensures the env vars are recognized
// Test invalid timeout (should use default)
os.Setenv("APTLY_ETCD_TIMEOUT", "invalid")
timeout := os.Getenv("APTLY_ETCD_TIMEOUT")
c.Assert(timeout, Equals, "invalid")
}
func (s *StorageSuite) TestIsTemporary(c *C) {
// Test nil error
c.Assert(isTemporary(nil), Equals, false)
// Test context deadline exceeded
c.Assert(isTemporary(context.DeadlineExceeded), Equals, true)
// Test timeout error
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Nanosecond)
defer cancel()
time.Sleep(10 * time.Millisecond)
<-ctx.Done()
c.Assert(isTemporary(ctx.Err()), Equals, true)
}
func (s *StorageSuite) TestApplyPrefix(c *C) {
// Test without temp prefix
storage := &EtcDStorage{}
key := []byte("test-key")
result := storage.applyPrefix(key)
c.Assert(result, DeepEquals, key)
// Test with temp prefix
storage.tmpPrefix = "temp123"
result = storage.applyPrefix(key)
expected := append([]byte("temp123/"), key...)
c.Assert(result, DeepEquals, expected)
}
// Mock test for retry logic
func (s *StorageSuite) TestGetRetryLogic(c *C) {
// This would require mocking etcd client, which is complex
// The test verifies the retry logic exists and compiles
// In production, this would be tested with integration tests
// Verify retry count
maxRetries := 3
c.Assert(maxRetries, Equals, 3)
// Verify backoff calculation
for i := 0; i < maxRetries; i++ {
backoff := time.Duration(i+1) * 100 * time.Millisecond
c.Assert(backoff >= 100*time.Millisecond, Equals, true)
c.Assert(backoff <= 300*time.Millisecond, Equals, true)
}
}
database/etcddb/transaction.go
+1 -5
View File
@@ -1,8 +1,6 @@
package etcddb
import (
"context"
"github.com/aptly-dev/aptly/database"
clientv3 "go.etcd.io/etcd/client/v3"
)
@@ -48,9 +46,7 @@ func (t *transaction) Commit() (err error) {
batchSize := 128
for i := 0; i < len(t.ops); i += batchSize {
ctx, cancel := context.WithTimeout(context.Background(), DefaultTimeout)
defer cancel()
txn := kv.Txn(ctx)
txn := kv.Txn(Ctx)
end := i + batchSize
if end > len(t.ops) {
end = len(t.ops)