Merge pull request #14380 from MinaProtocol/georgeee/orchestrator-on-aws

Use AWS on orchestrator
MinaProtocol · Oct 20, 2023 · 91d6adb · 91d6adb
2 parents 578a3f8 + 7909d94
commit 91d6adb
Show file tree

Hide file tree

Showing 5 changed files with 181 additions and 524 deletions.
diff --git a/src/app/itn_orchestrator/src/data.go b/src/app/itn_orchestrator/src/data.go
@@ -6,8 +6,8 @@ import (
 	"encoding/json"
 	"time"
 
-	"cloud.google.com/go/storage"
 	"github.com/Khan/genqlient/graphql"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
 	logging "github.com/ipfs/go-log/v2"
 )
 
@@ -38,9 +38,15 @@ type NodeEntry struct {
 	LastStatusCode  *int
 }
 
+type AwsContext struct {
+	Client     *s3.Client
+	BucketName *string
+	Prefix     string
+}
+
 type Config struct {
 	Ctx                context.Context
-	UptimeBucket       *storage.BucketHandle
+	AwsContext         AwsContext
 	Sk                 ed25519.PrivateKey
 	Log                logging.StandardLogger
 	MinaExec           string

diff --git a/src/app/itn_orchestrator/src/discovery.go b/src/app/itn_orchestrator/src/discovery.go
@@ -1,16 +1,17 @@
 package itn_orchestrator
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"strconv"
 	"strings"
 	"time"
 
-	"cloud.google.com/go/storage"
 	"github.com/Khan/genqlient/graphql"
-	"google.golang.org/api/iterator"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
 )
 
 func prefixByTime(t time.Time) string {
@@ -32,55 +33,78 @@ type DiscoveryParams struct {
 	Exactly            bool `json:"exactly,omitempty"`
 }
 
+func (awsctx AwsContext) ListObjects(ctx context.Context, startAfter string, continuationToken *string) (*s3.ListObjectsV2Output, error) {
+	return awsctx.Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
+		Bucket:            awsctx.BucketName,
+		Prefix:            aws.String(awsctx.Prefix),
+		StartAfter:        aws.String(awsctx.Prefix + "/" + startAfter),
+		ContinuationToken: continuationToken,
+	})
+}
+func (awsctx AwsContext) ReadObject(ctx context.Context, key *string) (*s3.GetObjectOutput, error) {
+	return awsctx.Client.GetObject(ctx, &s3.GetObjectInput{
+		Bucket: awsctx.BucketName,
+		Key:    key,
+	})
+}
+
 func discoverParticipantsDo(config Config, params DiscoveryParams, output func(NodeAddress)) error {
 	before := time.Now().Add(-time.Duration(params.OffsetMin) * time.Minute)
-	query := &storage.Query{StartOffset: prefixByTime(before)}
+	startAfter := prefixByTime(before)
 	log := config.Log
 	ctx := config.Ctx
-	it := config.UptimeBucket.Objects(ctx, query)
+
+	resp, err := config.AwsContext.ListObjects(ctx, startAfter, nil)
+	if err != nil {
+		return err
+	}
 	cache := make(map[NodeAddress]struct{})
 	for {
-		attrs, err := it.Next()
-		if err == iterator.Done {
-			break
-		}
-		if err != nil {
-			return err
-		}
-		r, err := config.UptimeBucket.Object(attrs.Name).NewReader(ctx)
-		if err != nil {
-			log.Errorf("Error reading submission %s: %v", attrs.Name, err)
-			continue
-		}
-		var meta MetaToBeSaved
-		d := json.NewDecoder(r)
-		err = d.Decode(&meta)
-		if err != nil {
-			log.Errorf("Error decoding submission %s: %v", attrs.Name, err)
-			continue
-		}
-		colonIx := strings.IndexRune(meta.RemoteAddr, ':')
-		if colonIx < 0 {
-			return fmt.Errorf("wrong remote address in submission %s: %s", attrs.Name, meta.RemoteAddr)
-		}
-		addr := NodeAddress(meta.RemoteAddr[:colonIx] + ":" + strconv.Itoa(int(meta.GraphqlControlPort)))
-		if _, has := cache[addr]; has {
-			continue
-		}
-		_, _, err = GetGqlClient(config, addr)
-		if err != nil {
-			log.Errorf("Error on auth for %s: %v", addr, err)
-			continue
-		}
-		if config.NodeData[addr].IsBlockProducer && params.NoBlockProducers {
-			continue
-		}
-		if !config.NodeData[addr].IsBlockProducer && params.OnlyBlockProducers {
-			continue
+		for _, obj := range resp.Contents {
+			name := *obj.Key
+			r, err := config.AwsContext.ReadObject(ctx, obj.Key)
+			if err != nil {
+				log.Errorf("Error reading submission %s: %v", name, err)
+				continue
+			}
+			var meta MetaToBeSaved
+			d := json.NewDecoder(r.Body)
+			err = d.Decode(&meta)
+			if err != nil {
+				log.Errorf("Error decoding submission %s: %v", name, err)
+				continue
+			}
+			colonIx := strings.IndexRune(meta.RemoteAddr, ':')
+			if colonIx < 0 {
+				return fmt.Errorf("wrong remote address in submission %s: %s", name, meta.RemoteAddr)
+			}
+			addr := NodeAddress(meta.RemoteAddr[:colonIx] + ":" + strconv.Itoa(int(meta.GraphqlControlPort)))
+			if _, has := cache[addr]; has {
+				continue
+			}
+			_, _, err = GetGqlClient(config, addr)
+			if err != nil {
+				log.Errorf("Error on auth for %s: %v", addr, err)
+				continue
+			}
+			if config.NodeData[addr].IsBlockProducer && params.NoBlockProducers {
+				continue
+			}
+			if !config.NodeData[addr].IsBlockProducer && params.OnlyBlockProducers {
+				continue
+			}
+			cache[addr] = struct{}{}
+			output(addr)
+			if params.Limit > 0 && len(cache) >= params.Limit {
+				break
+			}
 		}
-		cache[addr] = struct{}{}
-		output(addr)
-		if params.Limit > 0 && len(cache) >= params.Limit {
+		if resp.IsTruncated {
+			resp, err = config.AwsContext.ListObjects(ctx, startAfter, resp.NextContinuationToken)
+			if err != nil {
+				return err
+			}
+		} else {
 			break
 		}
 	}

diff --git a/src/app/itn_orchestrator/src/go.mod b/src/app/itn_orchestrator/src/go.mod
@@ -3,46 +3,43 @@ module itn_orchestrator
 go 1.20
 
 require (
-	cloud.google.com/go/storage v1.16.0
 	github.com/Khan/genqlient v0.5.0
+	github.com/aws/aws-sdk-go-v2 v1.21.1
+	github.com/aws/aws-sdk-go-v2/config v1.18.44
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.40.1
 	github.com/btcsuite/btcutil v1.0.2
 	github.com/ipfs/go-log/v2 v2.1.3
 	github.com/stretchr/testify v1.7.0
 	go.uber.org/zap v1.16.0
 	golang.org/x/crypto v0.0.0-20210921155107-089bfa567519
-	google.golang.org/api v0.49.0
 	itn_json_types v0.0.0
 )
 
 require (
-	cloud.google.com/go v0.84.0 // indirect
-	github.com/agnivade/levenshtein v1.1.1 // indirect
-	github.com/alexflint/go-arg v1.4.2 // indirect
-	github.com/alexflint/go-scalar v1.0.0 // indirect
+	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.4.14 // indirect
+	github.com/aws/aws-sdk-go-v2/credentials v1.13.42 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.13.12 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.42 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.36 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.3.44 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/v4a v1.1.5 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.9.15 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.1.37 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.36 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.15.5 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.15.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.17.2 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.23.1 // indirect
+	github.com/aws/smithy-go v1.15.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e // indirect
-	github.com/golang/protobuf v1.5.2 // indirect
-	github.com/googleapis/gax-go/v2 v2.0.5 // indirect
-	github.com/jstemmer/go-junit-report v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/vektah/gqlparser/v2 v2.4.5 // indirect
-	go.opencensus.io v0.23.0 // indirect
 	go.uber.org/atomic v1.7.0 // indirect
 	go.uber.org/multierr v1.6.0 // indirect
 	golang.org/x/lint v0.0.0-20210508222113-6edffad5e616 // indirect
-	golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3 // indirect
-	golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f // indirect
-	golang.org/x/oauth2 v0.0.0-20210615190721-d04028783cf1 // indirect
 	golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 // indirect
-	golang.org/x/text v0.3.7 // indirect
-	golang.org/x/tools v0.1.10 // indirect
-	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
-	google.golang.org/appengine v1.6.7 // indirect
-	google.golang.org/genproto v0.0.0-20210624174822-c5cf32407d0a // indirect
-	google.golang.org/grpc v1.38.0 // indirect
-	google.golang.org/protobuf v1.26.0 // indirect
-	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c // indirect
+	honnef.co/go/tools v0.0.1-2020.1.4 // indirect
 )
 
 replace itn_json_types => ../json_types