Skip to content

Commit 10d5c8c

Browse files
mxschmittclaude
mxschmitt
and
claude
authored
fix: Go code quality improvements (#301)
* chore: replace deprecated APIs and unmaintained packages - Replace io/ioutil with io (deprecated since Go 1.16) - Remove math/rand.Seed (auto-seeded since Go 1.20) - Use crypto/rand for secure random string generation - Migrate from streadway/amqp to rabbitmq/amqp091-go (official fork) - Replace k8s.io/utils/pointer with k8s.io/utils/ptr Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * fix: resource leaks from defer in loops and unclosed connections - Extract file processing to helper functions so defer executes properly in file-service and worker packages - Store and close AMQP connection on server shutdown in control-service The defer statement inside a loop doesn't close resources until the function returns, potentially exhausting file descriptors when processing many files. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * fix: propagate request context instead of using context.Background Use c.Request().Context() in HTTP handlers to properly propagate cancellation signals. This ensures that when a client disconnects, the server stops processing the request instead of continuing to make etcd calls unnecessarily. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * refactor: use sync.Map for worker reply channels Replace manual mutex + map pattern with sync.Map for storing worker reply channels. sync.Map is optimized for the concurrent access pattern used here: many reads with relatively few writes. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * fix: rename WorkerExectionOptions to WorkerExecutionOptions Fix typo in exported struct name (missing 'u' in Execution). Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * fix: typos in log/error messages - "could n ot" -> "could not" - "recursivelyt" -> "recursively" Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * fix: handle missing reply channel gracefully in Subscribe Add defensive check to avoid panic if reply channel is not found in sync.Map (shouldn't happen in normal flow, but prevents crash). Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * ci: add debug step on failure and timeout for health check - Add 2 minute timeout to wait-on health check - Add debug step that runs on failure to collect: - Pod status - Control service logs - Pod descriptions - RabbitMQ and etcd logs - Service status - Direct health endpoint check from within cluster Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * ci: reduce timeout for kubectl wait on worker pods - Decrease timeout from 10 minutes to 3 minutes for waiting on worker pods to be ready in CI workflow. * ci: fix debug step label selectors and add previous logs - Use deploy/control instead of label selectors - Add --previous flag to get logs from crashed container - Fix label selector to io.kompose.service Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * fix: amqp heartbeat parameter format for rabbitmq/amqp091-go The new library expects heartbeat as integer seconds (heartbeat=5), not duration string (heartbeat=5s). Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * ci: add file service logs to debug step Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * refactor: use slices.Contains for MIME type validation Replace verbose chain of != comparisons with slices.Contains for cleaner, more maintainable code. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * refactor: use slices.Contains in more places - workertypes.IsValid(): replace manual loop with slices.Contains - worker-java transformOutput(): simplify forbidden line filtering Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
1 parent 55b9eaf commit 10d5c8c

14 files changed

Lines changed: 157 additions & 110 deletions

File tree

‎.github/workflows/nodejs.yml‎

Lines changed: 25 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -49,11 +49,34 @@ jobs:
4949
FRONTEND_PORT=$(kubectl get svc frontend -o=jsonpath='{.spec.ports[0].nodePort}')
5050
FRONTEND_URL="http://127.0.0.1:$FRONTEND_PORT"
5151
echo "Host: $FRONTEND_URL"
52-
npx wait-on "$FRONTEND_URL/service/control/health"
53-
kubectl wait --timeout 10m --for=condition=ready pod -l role=worker
52+
npx wait-on --timeout 120000 "$FRONTEND_URL/service/control/health"
53+
kubectl wait --timeout 3m --for=condition=ready pod -l role=worker
5454
ROOT_TEST_URL=$FRONTEND_URL npm run test
5555
env:
5656
FLAKINESS_ACCESS_TOKEN: ${{ secrets.FLAKINESS_ACCESS_TOKEN }}
57+
- name: Debug on failure
58+
if: failure()
59+
run: |
60+
echo "=== Pod Status ==="
61+
kubectl get pods -o wide
62+
echo ""
63+
echo "=== Control Service Logs ==="
64+
kubectl logs deploy/control --tail=200 || true
65+
echo ""
66+
echo "=== Control Service Previous Logs (crashed) ==="
67+
kubectl logs deploy/control --previous --tail=200 || true
68+
echo ""
69+
echo "=== Control Service Describe ==="
70+
kubectl describe pod -l io.kompose.service=control || true
71+
echo ""
72+
echo "=== RabbitMQ Logs ==="
73+
kubectl logs deploy/rabbitmq --tail=50 || true
74+
echo ""
75+
echo "=== etcd Logs ==="
76+
kubectl logs deploy/etcd --tail=50 || true
77+
echo ""
78+
echo "=== File Service Logs ==="
79+
kubectl logs deploy/file --tail=100 || true
5780
- name: Upload playwright-report
5881
if: ${{ !cancelled() }}
5982
uses: actions/upload-artifact@v4

‎control-service/main.go‎

Lines changed: 24 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -2,10 +2,11 @@ package main
22

33
import (
44
"context"
5+
"crypto/rand"
56
"errors"
67
"fmt"
7-
"io/ioutil"
8-
"math/rand"
8+
"io"
9+
"math/big"
910
"net/http"
1011
"os"
1112
"os/signal"
@@ -21,7 +22,7 @@ import (
2122
"github.com/getsentry/sentry-go"
2223
sentryecho "github.com/getsentry/sentry-go/echo"
2324

24-
"github.com/streadway/amqp"
25+
amqp "github.com/rabbitmq/amqp091-go"
2526
clientv3 "go.etcd.io/etcd/client/v3"
2627
"k8s.io/client-go/kubernetes"
2728
"k8s.io/client-go/rest"
@@ -35,7 +36,6 @@ const (
3536
)
3637

3738
func init() {
38-
rand.Seed(time.Now().UTC().UnixNano())
3939
log.SetFormatter(&log.TextFormatter{
4040
TimestampFormat: time.StampMilli,
4141
})
@@ -46,7 +46,8 @@ type server struct {
4646

4747
etcdClient *clientv3.Client
4848

49-
amqpErrorChan chan *amqp.Error
49+
amqpConnection *amqp.Connection
50+
amqpErrorChan chan *amqp.Error
5051

5152
workers map[workertypes.WorkerLanguage]*Workers
5253
}
@@ -105,9 +106,10 @@ func newServer() (*server, error) {
105106
}
106107

107108
s := &server{
108-
etcdClient: etcdClient,
109-
amqpErrorChan: amqpErrorChan,
110-
workers: workersMap,
109+
etcdClient: etcdClient,
110+
amqpConnection: amqpConnection,
111+
amqpErrorChan: amqpErrorChan,
112+
workers: workersMap,
111113
}
112114

113115
s.initializeHttpServer()
@@ -216,8 +218,9 @@ func (s *server) handleRun(c echo.Context) error {
216218
}
217219

218220
func (s *server) handleShareGet(c echo.Context) error {
221+
ctx := c.Request().Context()
219222
id := c.Param("id")
220-
resp, err := s.etcdClient.Get(context.Background(), id)
223+
resp, err := s.etcdClient.Get(ctx, id)
221224
if err != nil {
222225
return fmt.Errorf("could not fetch share: %w", err)
223226
}
@@ -228,18 +231,19 @@ func (s *server) handleShareGet(c echo.Context) error {
228231
}
229232

230233
func (s *server) handleShareCreate(c echo.Context) error {
231-
code, err := ioutil.ReadAll(http.MaxBytesReader(c.Response().Writer, c.Request().Body, 1<<20))
234+
ctx := c.Request().Context()
235+
code, err := io.ReadAll(http.MaxBytesReader(c.Response().Writer, c.Request().Body, 1<<20))
232236
if err != nil {
233237
return fmt.Errorf("could not read request body: %w", err)
234238
}
235239
for retryCount := 0; retryCount <= 3; retryCount++ {
236240
id := generateRandomString(SNIPPET_ID_LENGTH)
237-
resp, err := s.etcdClient.Get(context.Background(), id)
241+
resp, err := s.etcdClient.Get(ctx, id)
238242
if err != nil {
239243
return fmt.Errorf("could not fetch share: %w", err)
240244
}
241245
if resp.Count == 0 {
242-
_, err = s.etcdClient.Put(context.Background(), id, string(code))
246+
_, err = s.etcdClient.Put(ctx, id, string(code))
243247
if err != nil {
244248
return fmt.Errorf("could not save share: %w", err)
245249
}
@@ -252,8 +256,9 @@ func (s *server) handleShareCreate(c echo.Context) error {
252256
}
253257

254258
func (s *server) handleHealth(c echo.Context) error {
259+
ctx := c.Request().Context()
255260
for _, endpoint := range s.etcdClient.Endpoints() {
256-
if _, err := s.etcdClient.Status(context.Background(), endpoint); err != nil {
261+
if _, err := s.etcdClient.Status(ctx, endpoint); err != nil {
257262
return fmt.Errorf("could not check etcd status: %w", err)
258263
}
259264
}
@@ -273,7 +278,9 @@ func (s *server) Stop() error {
273278
return fmt.Errorf("could not cleanup workers: %w", err)
274279
}
275280
}
276-
281+
if err := s.amqpConnection.Close(); err != nil {
282+
return fmt.Errorf("could not close amqp connection: %w", err)
283+
}
277284
return s.etcdClient.Close()
278285
}
279286

@@ -304,10 +311,11 @@ func main() {
304311
}
305312

306313
func generateRandomString(n int) string {
307-
var letterRunes = []rune("abcdefghijklmnopqrstuvpxyz1234567890")
314+
var letterRunes = []rune("abcdefghijklmnopqrstuvwxyz1234567890")
308315
b := make([]rune, n)
309316
for i := range b {
310-
b[i] = letterRunes[rand.Intn(len(letterRunes))]
317+
idx, _ := rand.Int(rand.Reader, big.NewInt(int64(len(letterRunes))))
318+
b[i] = letterRunes[idx.Int64()]
311319
}
312320
return string(b)
313321
}

‎control-service/workers.go‎

Lines changed: 19 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -11,12 +11,12 @@ import (
1111
log "github.com/sirupsen/logrus"
1212

1313
"github.com/google/uuid"
14-
"github.com/streadway/amqp"
14+
amqp "github.com/rabbitmq/amqp091-go"
1515
v1 "k8s.io/api/core/v1"
1616
"k8s.io/apimachinery/pkg/api/resource"
1717
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
1818
"k8s.io/client-go/kubernetes"
19-
"k8s.io/utils/pointer"
19+
"k8s.io/utils/ptr"
2020
)
2121

2222
type Workers struct {
@@ -25,14 +25,12 @@ type Workers struct {
2525
amqpReplyQueueName string
2626
amqpChannel *amqp.Channel
2727
k8ClientSet kubernetes.Interface
28-
repliesMu sync.Mutex
29-
replies map[string]chan *workertypes.WorkerResponsePayload
28+
replies sync.Map // map[string]chan *workertypes.WorkerResponsePayload
3029
}
3130

3231
func newWorkers(language workertypes.WorkerLanguage, workerCount int, k8ClientSet kubernetes.Interface, amqpChannel *amqp.Channel) (*Workers, error) {
3332
w := &Workers{
3433
language: language,
35-
replies: make(map[string]chan *workertypes.WorkerResponsePayload),
3634
k8ClientSet: k8ClientSet,
3735
amqpChannel: amqpChannel,
3836
workers: make(chan *Worker, workerCount),
@@ -75,13 +73,12 @@ func (w *Workers) consumeReplies() error {
7573
go func() {
7674
for msg := range msgs {
7775
log.Printf("received rpc callback, corr id: %v", msg.CorrelationId)
78-
w.repliesMu.Lock()
79-
replyChan, ok := w.replies[msg.CorrelationId]
80-
w.repliesMu.Unlock()
76+
value, ok := w.replies.Load(msg.CorrelationId)
8177
if !ok {
8278
log.Printf("no reply channel exists for worker %s", msg.CorrelationId)
8379
continue
8480
}
81+
replyChan := value.(chan *workertypes.WorkerResponsePayload)
8582
var reply *workertypes.WorkerResponsePayload
8683
if err := json.Unmarshal(msg.Body, &reply); err != nil {
8784
log.Printf("could not unmarshal reply json: %v", err)
@@ -132,9 +129,7 @@ func newWorker(workers *Workers) (*Worker, error) {
132129
language: workers.language,
133130
}
134131

135-
w.workers.repliesMu.Lock()
136-
w.workers.replies[w.id] = make(chan *workertypes.WorkerResponsePayload, 1)
137-
w.workers.repliesMu.Unlock()
132+
w.workers.replies.Store(w.id, make(chan *workertypes.WorkerResponsePayload, 1))
138133

139134
_, err := w.workers.amqpChannel.QueueDeclare(
140135
fmt.Sprintf("rpc_queue_%s", w.id), // name
@@ -167,8 +162,8 @@ func (w *Worker) createPod() error {
167162
},
168163
Spec: v1.PodSpec{
169164
RestartPolicy: v1.RestartPolicy(v1.RestartPolicyNever),
170-
AutomountServiceAccountToken: pointer.BoolPtr(false),
171-
EnableServiceLinks: pointer.BoolPtr(false),
165+
AutomountServiceAccountToken: ptr.To(false),
166+
EnableServiceLinks: ptr.To(false),
172167
Containers: []v1.Container{
173168
{
174169
Name: "worker",
@@ -181,7 +176,7 @@ func (w *Worker) createPod() error {
181176
},
182177
{
183178
Name: "AMQP_URL",
184-
Value: "amqp://rabbitmq:5672?heartbeat=5s",
179+
Value: "amqp://rabbitmq:5672?heartbeat=5",
185180
},
186181
{
187182
Name: "WORKER_HTTP_PROXY",
@@ -245,20 +240,21 @@ func (w *Worker) Publish(code string) error {
245240
func (w *Worker) Cleanup() error {
246241
if err := w.workers.k8ClientSet.CoreV1().Pods(K8_NAMESPACE_NAME).
247242
Delete(context.Background(), w.pod.Name, metav1.DeleteOptions{
248-
GracePeriodSeconds: pointer.Int64Ptr(0),
243+
GracePeriodSeconds: ptr.To(int64(0)),
249244
}); err != nil {
250245
return fmt.Errorf("could not delete pod: %w", err)
251246
}
252-
w.workers.repliesMu.Lock()
253-
delete(w.workers.replies, w.id)
254-
w.workers.repliesMu.Unlock()
255-
247+
w.workers.replies.Delete(w.id)
256248
return nil
257249
}
258250

259251
func (w *Worker) Subscribe() <-chan *workertypes.WorkerResponsePayload {
260-
w.workers.repliesMu.Lock()
261-
ch := w.workers.replies[w.id]
262-
w.workers.repliesMu.Unlock()
263-
return ch
252+
value, ok := w.workers.replies.Load(w.id)
253+
if !ok {
254+
// This shouldn't happen, but return a closed channel to avoid panic
255+
ch := make(chan *workertypes.WorkerResponsePayload)
256+
close(ch)
257+
return ch
258+
}
259+
return value.(chan *workertypes.WorkerResponsePayload)
264260
}

‎file-service/main.go‎

Lines changed: 52 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -5,11 +5,13 @@ import (
55
"context"
66
"fmt"
77
"io"
8+
"mime/multipart"
89
"net/http"
910
"net/url"
1011
"os"
1112
"os/signal"
1213
"path/filepath"
14+
"slices"
1315
"syscall"
1416
"time"
1517

@@ -34,6 +36,13 @@ type server struct {
3436

3537
const BUCKET_NAME = "file-uploads"
3638

39+
var allowedMimeTypes = []string{
40+
"application/pdf",
41+
"image/png",
42+
"video/webm",
43+
"application/zip",
44+
}
45+
3746
func newServer() (*server, error) {
3847
err := sentry.Init(sentry.ClientOptions{
3948
Dsn: os.Getenv("FILE_SERVICE_SENTRY_DSN"),
@@ -100,43 +109,56 @@ func (s *server) handleUploadImage(c echo.Context) error {
100109
outFiles := []publicFile{}
101110
for _, files := range c.Request().MultipartForm.File {
102111
for i := range files {
103-
file, err := files[i].Open()
104-
if err != nil {
105-
return fmt.Errorf("could not open file: %w", err)
106-
}
107-
fileContent, err := io.ReadAll(file)
108-
if err != nil {
109-
return fmt.Errorf("could not read file: %w", err)
110-
}
111-
defer file.Close()
112-
mimeType, err := filetype.Match(fileContent)
113-
if err != nil {
114-
return fmt.Errorf("could not detect mime-type: %w", err)
115-
}
116-
if mimeType.MIME.Value != "application/pdf" && mimeType.MIME.Value != "image/png" && mimeType.MIME.Value != "video/webm" && mimeType.MIME.Value != "application/zip" {
117-
return fmt.Errorf("not allowed mime-type (%s): %s", mimeType.MIME.Value, files[i].Filename)
118-
}
119-
fileExtension := filepath.Ext(files[i].Filename)
120-
objectName := uuid.New().String() + fileExtension
121-
if _, err := s.minioClient.PutObject(context.Background(), BUCKET_NAME, objectName, bytes.NewBuffer(fileContent), files[i].Size, minio.PutObjectOptions{
122-
ContentType: mimeType.MIME.Value,
123-
}); err != nil {
124-
return fmt.Errorf("could not put object: %w", err)
125-
}
126-
publicURL, err := s.minioClient.PresignedGetObject(context.Background(), BUCKET_NAME, objectName, time.Minute*10, url.Values{})
112+
pf, err := s.processUploadedFile(c.Request().Context(), files[i])
127113
if err != nil {
128-
return fmt.Errorf("could not generate public URL: %w", err)
114+
return err
129115
}
130-
outFiles = append(outFiles, publicFile{
131-
Extension: fileExtension,
132-
FileName: files[i].Filename,
133-
PublicURL: publicURL.EscapedPath() + "?" + publicURL.RawQuery,
134-
})
116+
outFiles = append(outFiles, pf)
135117
}
136118
}
137119
return c.JSON(http.StatusCreated, outFiles)
138120
}
139121

122+
func (s *server) processUploadedFile(ctx context.Context, fh *multipart.FileHeader) (publicFile, error) {
123+
file, err := fh.Open()
124+
if err != nil {
125+
return publicFile{}, fmt.Errorf("could not open file: %w", err)
126+
}
127+
defer file.Close()
128+
129+
fileContent, err := io.ReadAll(file)
130+
if err != nil {
131+
return publicFile{}, fmt.Errorf("could not read file: %w", err)
132+
}
133+
134+
mimeType, err := filetype.Match(fileContent)
135+
if err != nil {
136+
return publicFile{}, fmt.Errorf("could not detect mime-type: %w", err)
137+
}
138+
if !slices.Contains(allowedMimeTypes, mimeType.MIME.Value) {
139+
return publicFile{}, fmt.Errorf("not allowed mime-type (%s): %s", mimeType.MIME.Value, fh.Filename)
140+
}
141+
142+
fileExtension := filepath.Ext(fh.Filename)
143+
objectName := uuid.New().String() + fileExtension
144+
if _, err := s.minioClient.PutObject(ctx, BUCKET_NAME, objectName, bytes.NewBuffer(fileContent), fh.Size, minio.PutObjectOptions{
145+
ContentType: mimeType.MIME.Value,
146+
}); err != nil {
147+
return publicFile{}, fmt.Errorf("could not put object: %w", err)
148+
}
149+
150+
publicURL, err := s.minioClient.PresignedGetObject(ctx, BUCKET_NAME, objectName, time.Minute*10, url.Values{})
151+
if err != nil {
152+
return publicFile{}, fmt.Errorf("could not generate public URL: %w", err)
153+
}
154+
155+
return publicFile{
156+
Extension: fileExtension,
157+
FileName: fh.Filename,
158+
PublicURL: publicURL.EscapedPath() + "?" + publicURL.RawQuery,
159+
}, nil
160+
}
161+
140162
func (s *server) handleHealth(c echo.Context) error {
141163
return c.String(http.StatusOK, "OK")
142164
}

‎go.mod‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,8 +11,8 @@ require (
1111
github.com/h2non/filetype v1.1.3
1212
github.com/labstack/echo/v4 v4.15.0
1313
github.com/minio/minio-go/v7 v7.0.98
14+
github.com/rabbitmq/amqp091-go v1.10.0
1415
github.com/sirupsen/logrus v1.9.4
15-
github.com/streadway/amqp v1.1.0
1616
go.etcd.io/etcd/client/v3 v3.5.18
1717
k8s.io/api v0.35.0
1818
k8s.io/apimachinery v0.35.0

0 commit comments

Comments
 (0)