blob: 33c3e0cbde7f900d8aabd24fea6dfccf813aec6f [file] [log] [blame]
// Copyright 2023 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
//go:build linux || darwin
package gomote
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"io/fs"
"log"
"net/http"
"slices"
"sort"
"strings"
"time"
"cloud.google.com/go/storage"
"github.com/google/uuid"
buildbucketpb "go.chromium.org/luci/buildbucket/proto"
"go.chromium.org/luci/swarming/client/swarming"
swarmpb "go.chromium.org/luci/swarming/proto/api_v2"
"golang.org/x/build/buildlet"
"golang.org/x/build/internal"
"golang.org/x/build/internal/access"
"golang.org/x/build/internal/coordinator/remote"
"golang.org/x/build/internal/gomote/protos"
"golang.org/x/build/internal/rendezvous"
"golang.org/x/crypto/ssh"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
// expDisableGolangbuild disables the use of golangbuild during the swarming task gomote bootstrap process.
const expDisableGolangbuild = "disable-golang-build"
type rendezvousClient interface {
DeregisterInstance(ctx context.Context, id string)
HandleReverse(w http.ResponseWriter, r *http.Request)
RegisterInstance(ctx context.Context, id string, wait time.Duration)
WaitForInstance(ctx context.Context, id string) (buildlet.Client, error)
}
// BuildersClient is a partial interface of the buildbuicketpb.BuildersClient interface.
type BuildersClient interface {
GetBuilder(ctx context.Context, in *buildbucketpb.GetBuilderRequest, opts ...grpc.CallOption) (*buildbucketpb.BuilderItem, error)
ListBuilders(ctx context.Context, in *buildbucketpb.ListBuildersRequest, opts ...grpc.CallOption) (*buildbucketpb.ListBuildersResponse, error)
}
// SwarmingServer is a gomote server implementation which supports LUCI swarming bots.
type SwarmingServer struct {
// embed the unimplemented server.
protos.UnimplementedGomoteServiceServer
bucket bucketHandle
buildersClient BuildersClient
buildlets *remote.SessionPool
gceBucketName string
rendezvous rendezvousClient
sshCertificateAuthority ssh.Signer
swarmingClient swarming.Client
}
// NewSwarming creates a gomote server. If the rawCAPriKey is invalid, the program will exit.
func NewSwarming(rsp *remote.SessionPool, rawCAPriKey []byte, gomoteGCSBucket string, storageClient *storage.Client, rdv *rendezvous.Rendezvous, swarmClient swarming.Client, buildersClient buildbucketpb.BuildersClient) (*SwarmingServer, error) {
signer, err := ssh.ParsePrivateKey(rawCAPriKey)
if err != nil {
return nil, fmt.Errorf("unable to parse raw certificate authority private key into signer=%w", err)
}
return &SwarmingServer{
bucket: storageClient.Bucket(gomoteGCSBucket),
buildersClient: buildersClient,
buildlets: rsp,
gceBucketName: gomoteGCSBucket,
rendezvous: rdv,
sshCertificateAuthority: signer,
swarmingClient: swarmClient,
}, nil
}
// Authenticate will allow the caller to verify that they are properly authenticated and authorized to interact with the
// Service.
func (ss *SwarmingServer) Authenticate(ctx context.Context, req *protos.AuthenticateRequest) (*protos.AuthenticateResponse, error) {
_, err := access.IAPFromContext(ctx)
if err != nil {
log.Printf("Authenticate access.IAPFromContext(ctx) = nil, %s", err)
return nil, status.Errorf(codes.Unauthenticated, "request does not contain the required authentication")
}
return &protos.AuthenticateResponse{}, nil
}
// AddBootstrap adds the bootstrap version of Go to an instance and returns the URL for the bootstrap version. If no
// bootstrap version is defined then the returned version URL will be empty.
func (ss *SwarmingServer) AddBootstrap(ctx context.Context, req *protos.AddBootstrapRequest) (*protos.AddBootstrapResponse, error) {
creds, err := access.IAPFromContext(ctx)
if err != nil {
log.Printf("AddBootstrap access.IAPFromContext(ctx) = nil, %s", err)
return nil, status.Errorf(codes.Unauthenticated, "request does not contain the required authentication")
}
ses, bc, err := ss.sessionAndClient(ctx, req.GetGomoteId(), creds.ID)
if err != nil {
// the helper function returns meaningful GRPC error.
return nil, err
}
bs, err := ss.validBuilders(ctx)
if err != nil {
return nil, err
}
builder, ok := bs[ses.BuilderType]
if !ok {
return nil, status.Errorf(codes.Internal, "unable to determine builder definition")
}
cp, err := builderProperties(builder)
if err != nil {
log.Printf("AddBootstrap: bootstrap version not found for %s: %s", builder.GetId().GetBuilder(), err)
return &protos.AddBootstrapResponse{}, nil
}
if cp.BootstrapVersion == "latest" {
return &protos.AddBootstrapResponse{}, nil
}
var cipdPlatform string
for _, bd := range builder.GetConfig().GetDimensions() {
if !strings.HasPrefix(bd, "cipd_platform:") {
continue
}
var ok bool
_, cipdPlatform, ok = strings.Cut(bd, ":")
if !ok {
return nil, status.Errorf(codes.Internal, "unknown builder type")
}
break
}
goos, goarch, err := platformToGoValues(cipdPlatform)
if err != nil {
return nil, status.Errorf(codes.Internal, "unknown platform type")
}
url := fmt.Sprintf("https://storage.googleapis.com/go-builder-data/gobootstrap-%s-%s-go%s.tar.gz", goos, goarch, cp.BootstrapVersion)
if err = bc.PutTarFromURL(ctx, url, cp.BootstrapVersion); err != nil {
return nil, status.Errorf(codes.Internal, "unable to download bootstrap Go")
}
return &protos.AddBootstrapResponse{BootstrapGoUrl: url}, nil
}
// CreateInstance will create a gomote instance within a swarming task for the authenticated user.
func (ss *SwarmingServer) CreateInstance(req *protos.CreateInstanceRequest, stream protos.GomoteService_CreateInstanceServer) error {
creds, err := access.IAPFromContext(stream.Context())
if err != nil {
log.Printf("CreateInstance access.IAPFromContext(ctx) = nil, %s", err)
return status.Errorf(codes.Unauthenticated, "request does not contain the required authentication")
}
if req.GetBuilderType() == "" {
return status.Errorf(codes.InvalidArgument, "invalid builder type")
}
bs, err := ss.validBuilders(stream.Context())
if err != nil {
return err
}
builder, ok := bs[req.GetBuilderType()]
if !ok {
return status.Errorf(codes.InvalidArgument, "unknown builder type")
}
userName, err := emailToUser(creds.Email)
if err != nil {
return status.Errorf(codes.Internal, "invalid user email format")
}
type result struct {
buildletClient buildlet.Client
err error
}
rc := make(chan result, 1)
dimensions := make(map[string]string)
for _, bd := range builder.GetConfig().GetDimensions() {
k, v, ok := strings.Cut(bd, ":")
if ok {
dimensions[k] = v
} else {
log.Printf("failed dimension cut: %s", bd)
}
}
name := fmt.Sprintf("gomote-%s-%s", userName, uuid.NewString())
cp, err := builderProperties(builder)
if err != nil {
log.Printf("CreateInstance: builder configuration not found for %s: %s", builder.GetId().GetBuilder(), err)
return status.Errorf(codes.Internal, "invalid builder configuration")
}
useGolangbuild := !slices.Contains(req.GetExperimentOption(), expDisableGolangbuild)
go func() {
bc, err := ss.startNewSwarmingTask(stream.Context(), name, dimensions, cp, &SwarmOpts{}, useGolangbuild)
if err != nil {
log.Printf("startNewSwarmingTask() = %s", err)
}
rc <- result{bc, err}
}()
ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop()
for {
select {
case <-stream.Context().Done():
return status.Errorf(codes.DeadlineExceeded, "timed out waiting for gomote instance to be created")
case <-ticker.C:
err := stream.Send(&protos.CreateInstanceResponse{
Status: protos.CreateInstanceResponse_WAITING,
WaitersAhead: int64(0), // Not convinced querying for pending jobs is useful
})
if err != nil {
return status.Errorf(codes.Internal, "unable to stream result: %s", err)
}
case r := <-rc:
if r.err != nil {
log.Printf("error creating gomote buildlet instance=%s: %s", name, r.err)
return status.Errorf(codes.Internal, "gomote creation failed instance=%s", name)
}
gomoteID := ss.buildlets.AddSession(creds.ID, userName, req.GetBuilderType(), req.GetBuilderType(), r.buildletClient)
log.Printf("created buildlet %s for %s (%s)", gomoteID, userName, r.buildletClient.String())
session, err := ss.buildlets.Session(gomoteID)
if err != nil {
return status.Errorf(codes.Internal, "unable to query for gomote timeout") // this should never happen
}
wd, err := r.buildletClient.WorkDir(stream.Context())
if err != nil {
return status.Errorf(codes.Internal, "could not read working dir: %v", err)
}
err = stream.Send(&protos.CreateInstanceResponse{
Instance: &protos.Instance{
GomoteId: gomoteID,
BuilderType: req.GetBuilderType(),
HostType: "swarming task",
Expires: session.Expires.Unix(),
WorkingDir: wd,
},
Status: protos.CreateInstanceResponse_COMPLETE,
WaitersAhead: 0,
})
if err != nil {
return status.Errorf(codes.Internal, "unable to stream result: %s", err)
}
return nil
}
}
}
// DestroyInstance will destroy a gomote instance. It will ensure that the caller is authenticated and is the owner of the instance
// before it destroys the instance.
func (ss *SwarmingServer) DestroyInstance(ctx context.Context, req *protos.DestroyInstanceRequest) (*protos.DestroyInstanceResponse, error) {
creds, err := access.IAPFromContext(ctx)
if err != nil {
log.Printf("DestroyInstance access.IAPFromContext(ctx) = nil, %s", err)
return nil, status.Errorf(codes.Unauthenticated, "request does not contain the required authentication")
}
if req.GetGomoteId() == "" {
return nil, status.Errorf(codes.InvalidArgument, "invalid gomote ID")
}
_, err = ss.session(req.GetGomoteId(), creds.ID)
if err != nil {
// the helper function returns a meaningful GRPC error.
return nil, err
}
if err := ss.buildlets.DestroySession(req.GetGomoteId()); err != nil {
log.Printf("DestroyInstance remote.DestroySession(%s) = %s", req.GetGomoteId(), err)
return nil, status.Errorf(codes.Internal, "unable to destroy gomote instance")
}
// TODO(go.dev/issue/63819) consider destroying the bot after the task has ended.
return &protos.DestroyInstanceResponse{}, nil
}
// ExecuteCommand will execute a command on a gomote instance. The output from the command will be streamed back to the caller if the output is set.
func (ss *SwarmingServer) ExecuteCommand(req *protos.ExecuteCommandRequest, stream protos.GomoteService_ExecuteCommandServer) error {
creds, err := access.IAPFromContext(stream.Context())
if err != nil {
return status.Errorf(codes.Unauthenticated, "request does not contain the required authentication")
}
ses, bc, err := ss.sessionAndClient(stream.Context(), req.GetGomoteId(), creds.ID)
if err != nil {
// the helper function returns meaningful GRPC error.
return err
}
builderType := req.GetImitateHostType()
if builderType == "" {
builderType = ses.BuilderType
}
remoteErr, execErr := bc.Exec(stream.Context(), req.GetCommand(), buildlet.ExecOpts{
Dir: req.GetDirectory(),
SystemLevel: req.GetSystemLevel(),
Output: &streamWriter{writeFunc: func(p []byte) (int, error) {
err := stream.Send(&protos.ExecuteCommandResponse{
Output: p,
})
if err != nil {
return 0, fmt.Errorf("unable to send data=%w", err)
}
return len(p), nil
}},
Args: req.GetArgs(),
ExtraEnv: req.GetAppendEnvironment(),
Debug: req.GetDebug(),
Path: req.GetPath(),
})
if execErr != nil {
// there were system errors preventing the command from being started or seen to completion.
return status.Errorf(codes.Aborted, "unable to execute command: %s", execErr)
}
if remoteErr != nil {
// the command failed remotely
return status.Errorf(codes.Unknown, "command execution failed: %s", remoteErr)
}
return nil
}
// InstanceAlive will ensure that the gomote instance is still alive and will extend the timeout. The requester must be authenticated.
func (ss *SwarmingServer) InstanceAlive(ctx context.Context, req *protos.InstanceAliveRequest) (*protos.InstanceAliveResponse, error) {
creds, err := access.IAPFromContext(ctx)
if err != nil {
log.Printf("InstanceAlive access.IAPFromContext(ctx) = nil, %s", err)
return nil, status.Errorf(codes.Unauthenticated, "request does not contain the required authentication")
}
if req.GetGomoteId() == "" {
return nil, status.Errorf(codes.InvalidArgument, "invalid gomote ID")
}
_, err = ss.session(req.GetGomoteId(), creds.ID)
if err != nil {
// the helper function returns meaningful GRPC error.
return nil, err
}
if err := ss.buildlets.RenewTimeout(req.GetGomoteId()); err != nil {
return nil, status.Errorf(codes.Internal, "unable to renew timeout")
}
return &protos.InstanceAliveResponse{}, nil
}
// ListDirectory lists the contents of the directory on a gomote instance.
func (ss *SwarmingServer) ListDirectory(ctx context.Context, req *protos.ListDirectoryRequest) (*protos.ListDirectoryResponse, error) {
creds, err := access.IAPFromContext(ctx)
if err != nil {
return nil, status.Errorf(codes.Unauthenticated, "request does not contain the required authentication")
}
if req.GetGomoteId() == "" || req.GetDirectory() == "" {
return nil, status.Errorf(codes.InvalidArgument, "invalid arguments")
}
_, bc, err := ss.sessionAndClient(ctx, req.GetGomoteId(), creds.ID)
if err != nil {
// the helper function returns meaningful GRPC error.
return nil, err
}
opt := buildlet.ListDirOpts{
Recursive: req.GetRecursive(),
Digest: req.GetDigest(),
Skip: req.GetSkipFiles(),
}
var entries []string
if err = bc.ListDir(context.Background(), req.GetDirectory(), opt, func(bi buildlet.DirEntry) {
entries = append(entries, bi.String())
}); err != nil {
return nil, status.Errorf(codes.Unimplemented, "method ListDirectory not implemented")
}
return &protos.ListDirectoryResponse{
Entries: entries,
}, nil
}
const (
// golangbuildModeAll is golangbuild's MODE_ALL mode that
// builds and tests the project all within the same build.
//
// See https://source.chromium.org/chromium/infra/infra/+/main:go/src/infra/experimental/golangbuild/golangbuildpb/params.proto;l=148-149;drc=4e874bfb4ff7ff0620940712983ca82e8ea81028.
golangbuildModeAll = 0
// golangbuildPerfMode is golangbuild's MODE_PERF that
// runs performance tests.
//
// See https://source.chromium.org/chromium/infra/infra/+/main:go/src/infra/experimental/golangbuild/golangbuildpb/params.proto;l=174-177;drc=fdea4abccf8447808d4e702c8d09fdd20fd81acb.
golangbuildPerfMode = 4
)
func (ss *SwarmingServer) validBuilders(ctx context.Context) (map[string]*buildbucketpb.BuilderItem, error) {
listBuilders := func(bucket string) ([]*buildbucketpb.BuilderItem, error) {
var builders []*buildbucketpb.BuilderItem
var nextToken string
for {
buildersResp, err := ss.buildersClient.ListBuilders(ctx, &buildbucketpb.ListBuildersRequest{
Project: "golang",
Bucket: bucket,
PageSize: 1000,
PageToken: nextToken,
})
if err != nil {
return nil, err
}
builders = append(builders, buildersResp.GetBuilders()...)
if tok := buildersResp.GetNextPageToken(); tok != "" {
nextToken = tok
continue
}
return builders, nil
}
}
// list all the valid builders in ci-workers
builderBucket := "ci-workers"
builderResponse, err := listBuilders(builderBucket)
if err != nil {
log.Printf("buildersClient.ListBuilders(ctx, %s) = nil, %s", builderBucket, err)
return nil, status.Errorf(codes.Internal, "unable to query for builders")
}
builders := make(map[string]*buildbucketpb.BuilderItem)
for _, builder := range builderResponse {
bID := builder.GetId()
if bID == nil {
continue
}
name := bID.GetBuilder()
if !strings.HasSuffix(name, "-test_only") {
continue
}
builders[strings.TrimSuffix(name, "-test_only")] = builder
}
// list all the valid builders in ci
builderBucket = "ci"
builderResponse, err = listBuilders(builderBucket)
if err != nil {
log.Printf("buildersClient.ListBuilders(ctx, %s) = nil, %s", builderBucket, err)
return nil, status.Errorf(codes.Internal, "unable to query for builders")
}
for _, builder := range builderResponse {
bID := builder.GetId()
if bID == nil {
continue
}
name := bID.GetBuilder()
if _, ok := builders[name]; ok {
// should not happen
continue
}
config, err := builderProperties(builder)
if err != nil || !slices.Contains([]int{golangbuildModeAll, golangbuildPerfMode}, config.Mode) {
continue
}
builders[name] = builder
}
return builders, nil
}
// ListSwarmingBuilders lists all of the swarming builders which run for the Go master or release branches. The requester must be authenticated.
func (ss *SwarmingServer) ListSwarmingBuilders(ctx context.Context, req *protos.ListSwarmingBuildersRequest) (*protos.ListSwarmingBuildersResponse, error) {
_, err := access.IAPFromContext(ctx)
if err != nil {
log.Printf("ListSwarmingInstances access.IAPFromContext(ctx) = nil, %s", err)
return nil, status.Errorf(codes.Unauthenticated, "request does not contain the required authentication")
}
bs, err := ss.validBuilders(ctx)
if err != nil {
return nil, err
}
var builders []string
for builder, _ := range bs {
builders = append(builders, builder)
}
sort.Strings(builders)
return &protos.ListSwarmingBuildersResponse{Builders: builders}, nil
}
// ListInstances will list the gomote instances owned by the requester. The requester must be authenticated.
func (ss *SwarmingServer) ListInstances(ctx context.Context, req *protos.ListInstancesRequest) (*protos.ListInstancesResponse, error) {
creds, err := access.IAPFromContext(ctx)
if err != nil {
log.Printf("ListInstances access.IAPFromContext(ctx) = nil, %s", err)
return nil, status.Errorf(codes.Unauthenticated, "request does not contain the required authentication")
}
res := &protos.ListInstancesResponse{}
for _, s := range ss.buildlets.List() {
if s.OwnerID != creds.ID {
continue
}
res.Instances = append(res.Instances, &protos.Instance{
GomoteId: s.ID,
BuilderType: s.BuilderType,
HostType: s.HostType,
Expires: s.Expires.Unix(),
})
}
return res, nil
}
// ReadTGZToURL retrieves a directory from the gomote instance and writes the file to GCS. It returns a signed URL which the caller uses
// to read the file from GCS.
func (ss *SwarmingServer) ReadTGZToURL(ctx context.Context, req *protos.ReadTGZToURLRequest) (*protos.ReadTGZToURLResponse, error) {
creds, err := access.IAPFromContext(ctx)
if err != nil {
return nil, status.Errorf(codes.Unauthenticated, "request does not contain the required authentication")
}
_, bc, err := ss.sessionAndClient(ctx, req.GetGomoteId(), creds.ID)
if err != nil {
// the helper function returns meaningful GRPC error.
return nil, err
}
tgz, err := bc.GetTar(ctx, req.GetDirectory())
if err != nil {
return nil, status.Errorf(codes.Aborted, "unable to retrieve tar from gomote instance: %s", err)
}
defer tgz.Close()
objectName := uuid.NewString()
objectHandle := ss.bucket.Object(objectName)
// A context for writes is used to ensure we can cancel the context if a
// problem is encountered while writing to the object store. The API documentation
// states that the context should be canceled to stop writing without saving the data.
writeCtx, cancel := context.WithCancel(ctx)
tgzWriter := objectHandle.NewWriter(writeCtx)
defer cancel()
if _, err = io.Copy(tgzWriter, tgz); err != nil {
return nil, status.Errorf(codes.Aborted, "unable to stream tar.gz: %s", err)
}
// when close is called, the object is stored in the bucket.
if err := tgzWriter.Close(); err != nil {
return nil, status.Errorf(codes.Aborted, "unable to store object: %s", err)
}
url, err := ss.signURLForDownload(objectName)
if err != nil {
return nil, status.Errorf(codes.Internal, "unable to create signed URL for download: %s", err)
}
return &protos.ReadTGZToURLResponse{
Url: url,
}, nil
}
// RemoveFiles removes files or directories from the gomote instance.
func (ss *SwarmingServer) RemoveFiles(ctx context.Context, req *protos.RemoveFilesRequest) (*protos.RemoveFilesResponse, error) {
creds, err := access.IAPFromContext(ctx)
if err != nil {
log.Printf("RemoveFiles access.IAPFromContext(ctx) = nil, %s", err)
return nil, status.Errorf(codes.Unauthenticated, "request does not contain the required authentication")
}
// TODO(go.dev/issue/48742) consider what additional path validation should be implemented.
if req.GetGomoteId() == "" || len(req.GetPaths()) == 0 {
return nil, status.Errorf(codes.InvalidArgument, "invalid arguments")
}
_, bc, err := ss.sessionAndClient(ctx, req.GetGomoteId(), creds.ID)
if err != nil {
// the helper function returns meaningful GRPC error.
return nil, err
}
if err := bc.RemoveAll(ctx, req.GetPaths()...); err != nil {
log.Printf("RemoveFiles buildletClient.RemoveAll(ctx, %q) = %s", req.GetPaths(), err)
return nil, status.Errorf(codes.Unknown, "unable to remove files")
}
return &protos.RemoveFilesResponse{}, nil
}
// SignSSHKey signs the public SSH key with a certificate. The signed public SSH key is intended for use with the gomote service SSH
// server. It will be signed by the certificate authority of the server and will restrict access to the gomote instance that it was
// signed for.
func (ss *SwarmingServer) SignSSHKey(ctx context.Context, req *protos.SignSSHKeyRequest) (*protos.SignSSHKeyResponse, error) {
creds, err := access.IAPFromContext(ctx)
if err != nil {
return nil, status.Errorf(codes.Unauthenticated, "request does not contain the required authentication")
}
session, err := ss.session(req.GetGomoteId(), creds.ID)
if err != nil {
// the helper function returns meaningful GRPC error.
return nil, err
}
signedPublicKey, err := remote.SignPublicSSHKey(ctx, ss.sshCertificateAuthority, req.GetPublicSshKey(), session.ID, session.OwnerID, 5*time.Minute)
if err != nil {
return nil, status.Errorf(codes.InvalidArgument, "unable to sign ssh key")
}
return &protos.SignSSHKeyResponse{
SignedPublicSshKey: signedPublicKey,
}, nil
}
// UploadFile creates a URL and a set of HTTP post fields which are used to upload a file to a staging GCS bucket. Uploaded files are made available to the
// gomote instances via a subsequent call to one of the WriteFromURL endpoints.
func (ss *SwarmingServer) UploadFile(ctx context.Context, req *protos.UploadFileRequest) (*protos.UploadFileResponse, error) {
_, err := access.IAPFromContext(ctx)
if err != nil {
return nil, status.Errorf(codes.Unauthenticated, "request does not contain the required authentication")
}
objectName := uuid.NewString()
url, fields, err := ss.signURLForUpload(objectName)
if err != nil {
log.Printf("unable to create signed URL: %s", err)
return nil, status.Errorf(codes.Internal, "unable to create signed url")
}
return &protos.UploadFileResponse{
Url: url,
Fields: fields,
ObjectName: objectName,
}, nil
}
// signURLForUpload generates a signed URL and a set of http Post fields to be used to upload an object to GCS without authenticating.
func (ss *SwarmingServer) signURLForUpload(object string) (url string, fields map[string]string, err error) {
if object == "" {
return "", nil, errors.New("invalid object name")
}
pv4, err := ss.bucket.GenerateSignedPostPolicyV4(object, &storage.PostPolicyV4Options{
Expires: time.Now().Add(10 * time.Minute),
Insecure: false,
})
if err != nil {
return "", nil, fmt.Errorf("unable to generate signed url: %w", err)
}
return pv4.URL, pv4.Fields, nil
}
// WriteFileFromURL initiates an HTTP request to the passed in URL and streams the contents of the request to the gomote instance.
func (ss *SwarmingServer) WriteFileFromURL(ctx context.Context, req *protos.WriteFileFromURLRequest) (*protos.WriteFileFromURLResponse, error) {
creds, err := access.IAPFromContext(ctx)
if err != nil {
log.Printf("WriteTGZFromURL access.IAPFromContext(ctx) = nil, %s", err)
return nil, status.Errorf(codes.Unauthenticated, "request does not contain the required authentication")
}
_, bc, err := ss.sessionAndClient(ctx, req.GetGomoteId(), creds.ID)
if err != nil {
// the helper function returns meaningful GRPC error.
return nil, err
}
var rc io.ReadCloser
// objects stored in the gomote staging bucket are only accessible when you have been granted explicit permissions. A builder
// requires a signed URL in order to access objects stored in the gomote staging bucket.
if onObjectStore(ss.gceBucketName, req.GetUrl()) {
object, err := objectFromURL(ss.gceBucketName, req.GetUrl())
if err != nil {
return nil, status.Errorf(codes.InvalidArgument, "invalid object URL")
}
rc, err = ss.bucket.Object(object).NewReader(ctx)
if err != nil {
return nil, status.Errorf(codes.Internal, "unable to create object reader: %s", err)
}
} else {
httpRequest, err := http.NewRequestWithContext(ctx, http.MethodGet, req.GetUrl(), nil)
if err != nil {
log.Printf("gomote: unable to create HTTP request: %s", err)
return nil, status.Errorf(codes.Internal, "unable to create HTTP request")
}
// TODO(amedee) find sane client defaults, possibly rely on context timeout in request.
client := &http.Client{
Timeout: 30 * time.Second,
Transport: &http.Transport{
TLSHandshakeTimeout: 5 * time.Second,
},
}
resp, err := client.Do(httpRequest)
if err != nil {
return nil, status.Errorf(codes.Aborted, "failed to get file from URL: %s", err)
}
if resp.StatusCode != http.StatusOK {
return nil, status.Errorf(codes.Aborted, "unable to get file from %q: response code: %d", req.GetUrl(), resp.StatusCode)
}
rc = resp.Body
}
defer rc.Close()
if err := bc.Put(ctx, rc, req.GetFilename(), fs.FileMode(req.GetMode())); err != nil {
return nil, status.Errorf(codes.Aborted, "failed to send the file to the gomote instance: %s", err)
}
return &protos.WriteFileFromURLResponse{}, nil
}
// WriteTGZFromURL will instruct the gomote instance to download the tar.gz from the provided URL. The tar.gz file will be unpacked in the work directory
// relative to the directory provided.
func (ss *SwarmingServer) WriteTGZFromURL(ctx context.Context, req *protos.WriteTGZFromURLRequest) (*protos.WriteTGZFromURLResponse, error) {
creds, err := access.IAPFromContext(ctx)
if err != nil {
log.Printf("WriteTGZFromURL access.IAPFromContext(ctx) = nil, %s", err)
return nil, status.Errorf(codes.Unauthenticated, "request does not contain the required authentication")
}
if req.GetGomoteId() == "" {
return nil, status.Errorf(codes.InvalidArgument, "invalid gomote ID")
}
if req.GetUrl() == "" {
return nil, status.Errorf(codes.InvalidArgument, "missing URL")
}
_, bc, err := ss.sessionAndClient(ctx, req.GetGomoteId(), creds.ID)
if err != nil {
// the helper function returns meaningful GRPC error.
return nil, err
}
url := req.GetUrl()
if onObjectStore(ss.gceBucketName, url) {
object, err := objectFromURL(ss.gceBucketName, url)
if err != nil {
return nil, status.Errorf(codes.InvalidArgument, "invalid URL")
}
url, err = ss.signURLForDownload(object)
if err != nil {
return nil, status.Errorf(codes.Aborted, "unable to sign url for download: %s", err)
}
}
if err := bc.PutTarFromURL(ctx, url, req.GetDirectory()); err != nil {
return nil, status.Errorf(codes.FailedPrecondition, "unable to write tar.gz: %s", err)
}
return &protos.WriteTGZFromURLResponse{}, nil
}
// session is a helper function that retrieves a session associated with the gomoteID and ownerID.
func (ss *SwarmingServer) session(gomoteID, ownerID string) (*remote.Session, error) {
session, err := ss.buildlets.Session(gomoteID)
if err != nil {
return nil, status.Errorf(codes.NotFound, "specified gomote instance does not exist")
}
if session.OwnerID != ownerID {
return nil, status.Errorf(codes.PermissionDenied, "not allowed to modify this gomote session")
}
return session, nil
}
// sessionAndClient is a helper function that retrieves a session and buildlet client for the
// associated gomoteID and ownerID. The gomote instance timeout is renewed if the gomote id and owner id
// are valid.
func (ss *SwarmingServer) sessionAndClient(ctx context.Context, gomoteID, ownerID string) (*remote.Session, buildlet.Client, error) {
session, err := ss.session(gomoteID, ownerID)
if err != nil {
return nil, nil, err
}
bc, err := ss.buildlets.BuildletClient(gomoteID)
if err != nil {
return nil, nil, status.Errorf(codes.NotFound, "specified gomote instance does not exist")
}
if err := ss.buildlets.KeepAlive(ctx, gomoteID); err != nil {
log.Printf("gomote: unable to keep alive %s: %s", gomoteID, err)
}
return session, bc, nil
}
// signURLForDownload generates a signed URL and fields to be used to upload an object to GCS without authenticating.
func (ss *SwarmingServer) signURLForDownload(object string) (url string, err error) {
url, err = ss.bucket.SignedURL(object, &storage.SignedURLOptions{
Expires: time.Now().Add(10 * time.Minute),
Method: http.MethodGet,
Scheme: storage.SigningSchemeV4,
})
if err != nil {
return "", fmt.Errorf("unable to generate signed url: %w", err)
}
return url, err
}
// SwarmOpts provides additional options for swarming task creation.
type SwarmOpts struct {
// OnInstanceRequested optionally specifies a hook to run synchronously
// after the computeService.Instances.Insert call, but before
// waiting for its operation to proceed.
OnInstanceRequested func()
// OnInstanceCreated optionally specifies a hook to run synchronously
// after the instance operation succeeds.
OnInstanceCreated func()
// OnInstanceRegistration optionally specifies a hook to run synchronously
// after the instance has been registered in rendezvous.
OnInstanceRegistration func()
}
// startNewSwarmingTask starts a new swarming task on a bot with the buildlet
// running on it. It returns a buildlet client configured to speak to it.
// The request will last as long as the lifetime of the context. The dimensions
// are a set of key value pairs used to describe what instance type to create.
func (ss *SwarmingServer) startNewSwarmingTask(ctx context.Context, name string, dimensions map[string]string, properties *configProperties, opts *SwarmOpts, useGolangbuild bool) (buildlet.Client, error) {
ss.rendezvous.RegisterInstance(ctx, name, 10*time.Minute)
condRun(opts.OnInstanceRegistration)
taskID, err := ss.newSwarmingTask(ctx, name, dimensions, properties, opts, useGolangbuild)
if err != nil {
ss.rendezvous.DeregisterInstance(ctx, name)
return nil, err
}
log.Printf("gomote: swarming task requested name=%s taskID=%s", name, taskID)
condRun(opts.OnInstanceRequested)
queryCtx, cancel := context.WithTimeout(ctx, 20*time.Minute)
defer cancel()
var taskUp bool
tryThenPeriodicallyDo(queryCtx, 5*time.Second, func(ctx context.Context, _ time.Time) {
resp, err := ss.swarmingClient.TaskResult(ctx, taskID, &swarming.TaskResultFields{WithPerf: false})
if err != nil {
log.Printf("gomote: unable to query for swarming task state: name=%s taskID=%s %s", name, taskID, err)
return
}
switch taskState := resp.GetState(); taskState {
case swarmpb.TaskState_COMPLETED, swarmpb.TaskState_RUNNING:
taskUp = true
cancel()
case swarmpb.TaskState_EXPIRED, swarmpb.TaskState_INVALID, swarmpb.TaskState_BOT_DIED, swarmpb.TaskState_CANCELED,
swarmpb.TaskState_CLIENT_ERROR, swarmpb.TaskState_KILLED, swarmpb.TaskState_NO_RESOURCE, swarmpb.TaskState_TIMED_OUT:
log.Printf("gomote: swarming task creation failed name=%s state=%s", name, taskState)
cancel()
case swarmpb.TaskState_PENDING:
// continue waiting
default:
log.Printf("gomote: unexpected swarming task state for %s: %s", name, taskState)
}
})
if !taskUp {
ss.rendezvous.DeregisterInstance(ctx, name)
return nil, fmt.Errorf("unable to create swarming task name=%s taskID=%s", name, taskID)
}
condRun(opts.OnInstanceCreated)
bc, err := ss.waitForInstanceOrFailure(ctx, taskID, name)
if err != nil {
ss.rendezvous.DeregisterInstance(ctx, name)
return nil, err
}
return bc, nil
}
// waitForInstanceOrFailure waits for either the swarming task to enter a failed state or the successful connection from
// a buildlet client.
func (ss *SwarmingServer) waitForInstanceOrFailure(ctx context.Context, taskID, name string) (buildlet.Client, error) {
queryCtx, cancel := context.WithTimeout(ctx, 25*time.Minute)
checkForTaskFailure := func(pollCtx context.Context) <-chan error {
errCh := make(chan error, 1)
go func() {
internal.PeriodicallyDo(pollCtx, 10*time.Second, func(ctx context.Context, _ time.Time) {
resp, err := ss.swarmingClient.TaskResult(ctx, taskID, &swarming.TaskResultFields{WithPerf: false})
if err != nil {
log.Printf("gomote: unable to query for swarming task state: name=%s taskID=%s %s", name, taskID, err)
return
}
switch taskState := resp.GetState(); taskState {
case swarmpb.TaskState_RUNNING:
// expected
case swarmpb.TaskState_EXPIRED, swarmpb.TaskState_INVALID, swarmpb.TaskState_BOT_DIED, swarmpb.TaskState_CANCELED,
swarmpb.TaskState_CLIENT_ERROR, swarmpb.TaskState_KILLED, swarmpb.TaskState_NO_RESOURCE, swarmpb.TaskState_TIMED_OUT, swarmpb.TaskState_COMPLETED:
errCh <- fmt.Errorf("swarming task creation failed name=%s state=%s", name, taskState)
default:
log.Printf("gomote: unexpected swarming task state for %s: %s", name, taskState)
}
})
}()
return errCh
}
type result struct {
err error
bc buildlet.Client
}
getConn := func(waitCtx context.Context) <-chan *result {
ch := make(chan *result, 1)
go func() {
bc, err := ss.rendezvous.WaitForInstance(waitCtx, name)
if err != nil {
ss.rendezvous.DeregisterInstance(ctx, name)
}
ch <- &result{err: err, bc: bc}
}()
return ch
}
statusChan := checkForTaskFailure(queryCtx)
resChan := getConn(queryCtx)
select {
case err := <-statusChan:
cancel()
ss.rendezvous.DeregisterInstance(ctx, name)
log.Printf("gomote: failed waiting for task to run %q: %s", name, err)
return nil, err
case r := <-resChan:
cancel()
if r.err != nil {
log.Printf("gomote: failed to establish connection %q: %s", name, r.err)
return nil, r.err
}
return r.bc, r.err
}
}
func buildletStartup(goos, goarch string) string {
cmd := `import urllib.request
import sys
import platform
import subprocess
import os
import stat
def add_os_file_ext(filename):
if sys.platform == "win32":
return filename+".exe"
return filename
def sep():
if sys.platform == "win32":
return "\\"
else:
return "/"
def delete_if_exists(file_path):
if os.path.exists(file_path):
os.remove(file_path)
def make_executable(file_path):
if sys.platform != "win32":
st = os.stat(file_path)
os.chmod(file_path, st.st_mode | stat.S_IEXEC)
if __name__ == "__main__":
buildlet_name = add_os_file_ext("buildlet")
delete_if_exists(buildlet_name)
urllib.request.urlretrieve("https://storage.googleapis.com/go-builder-data/buildlet.%s-%s", buildlet_name)
make_executable(os.getcwd() + sep() + buildlet_name)
buildlet_name = "."+sep()+buildlet_name
subprocess.run([buildlet_name, "--coordinator=gomotessh.golang.org:443", "--reverse-type=swarming-task", "-swarming-bot", "-halt=false"], shell=False, env=os.environ.copy())
`
return fmt.Sprintf(cmd, goos, goarch)
}
func createStringPairs(m map[string]string) []*swarmpb.StringPair {
dims := make([]*swarmpb.StringPair, 0, len(m))
for k, v := range m {
dims = append(dims, &swarmpb.StringPair{
Key: k,
Value: v,
})
}
return dims
}
func platformToGoValues(platform string) (goos string, goarch string, err error) {
goos, goarch, ok := strings.Cut(platform, "-")
if !ok {
return "", "", fmt.Errorf("cipd_platform not in proper format=%s", platform)
}
if goos == "Mac" || goos == "mac" {
goos = "darwin"
}
if goarch == "armv6l" {
goarch = "arm"
}
return goos, goarch, nil
}
func (ss *SwarmingServer) newSwarmingTask(ctx context.Context, name string, dimensions map[string]string, properties *configProperties, opts *SwarmOpts, useGolangbuild bool) (string, error) {
if useGolangbuild {
return ss.newSwarmingTaskWithGolangbuild(ctx, name, dimensions, properties, opts)
}
cipdPlatform, ok := dimensions["cipd_platform"]
if !ok {
return "", fmt.Errorf("dimensions require cipd_platform: instance=%s", name)
}
goos, goarch, err := platformToGoValues(cipdPlatform)
if err != nil {
return "", err
}
packages := []*swarmpb.CipdPackage{
{Path: "tools/bin", PackageName: "infra/tools/luci-auth/" + cipdPlatform, Version: "latest"},
{Path: "tools/bootstrap-go", PackageName: "golang/bootstrap-go/" + cipdPlatform, Version: properties.BootstrapVersion},
}
pythonBin := "python3"
switch goos {
case "darwin":
pythonBin = `tools/bin/python3`
packages = append(packages,
&swarmpb.CipdPackage{Path: "tools/bin", PackageName: "infra/tools/mac_toolchain/" + cipdPlatform, Version: "latest"},
&swarmpb.CipdPackage{Path: "tools", PackageName: "infra/3pp/tools/cpython3/" + cipdPlatform, Version: "latest"})
case "windows":
pythonBin = `tools\bin\python3.exe`
packages = append(packages, &swarmpb.CipdPackage{Path: "tools", PackageName: "infra/3pp/tools/cpython3/" + cipdPlatform, Version: "latest"})
}
req := &swarmpb.NewTaskRequest{
Name: name,
Priority: 20, // 30 is the priority for builds
ServiceAccount: "coordinator-builder@golang-ci-luci.iam.gserviceaccount.com",
TaskSlices: []*swarmpb.TaskSlice{
&swarmpb.TaskSlice{
Properties: &swarmpb.TaskProperties{
CipdInput: &swarmpb.CipdInput{
Packages: packages,
},
EnvPrefixes: []*swarmpb.StringListPair{
{Key: "PATH", Value: []string{"tools/bin", "go/bin"}},
{Key: "GOROOT_BOOTSTRAP", Value: []string{"tools/bootstrap-go"}},
{Key: "GOPATH", Value: []string{"gopath"}},
},
Command: []string{pythonBin, "-c", buildletStartup(goos, goarch)},
Dimensions: createStringPairs(dimensions),
Env: []*swarmpb.StringPair{
&swarmpb.StringPair{
Key: "GOMOTEID",
Value: name,
},
},
ExecutionTimeoutSecs: 86400,
},
ExpirationSecs: 86400,
WaitForCapacity: false,
},
},
Tags: []string{"golang_mode:gomote"},
Realm: "golang:ci",
}
taskMD, err := ss.swarmingClient.NewTask(ctx, req)
if err != nil {
log.Printf("gomote: swarming task creation failed name=%s: %s", name, err)
return "", fmt.Errorf("unable to start task: %w", err)
}
log.Printf("gomote: task created: id=%s https://chromium-swarm.appspot.com/task?id=%s", taskMD.TaskId, taskMD.TaskId)
return taskMD.TaskId, nil
}
func golangbuildStartup(goos, goarch, golangbuildBin string) string {
cmd := `import urllib.request
import sys
import platform
import subprocess
import os
import stat
def make_executable(file_path):
if sys.platform != "win32":
st = os.stat(file_path)
os.chmod(file_path, st.st_mode | stat.S_IEXEC)
if __name__ == "__main__":
ext = ""
if sys.platform == "win32":
ext = ".exe"
sep = "/"
if sys.platform == "win32":
sep = "\\\\"
buildlet_file = "buildlet" + ext
buildlet_path = "." + sep + buildlet_file
if os.path.exists(buildlet_path):
os.remove(buildlet_path)
urllib.request.urlretrieve("https://storage.googleapis.com/go-builder-data/buildlet.%s-%s", buildlet_path)
make_executable(os.getcwd() + sep + buildlet_file)
subprocess.run(["%s", buildlet_path, "--coordinator=gomotessh.golang.org:443", "--reverse-type=swarming-task", "-swarming-bot", "-halt=false"], shell=False, env=os.environ.copy())
`
return fmt.Sprintf(cmd, goos, goarch, golangbuildBin)
}
func (ss *SwarmingServer) newSwarmingTaskWithGolangbuild(ctx context.Context, name string, dimensions map[string]string, properties *configProperties, opts *SwarmOpts) (string, error) {
log.Printf("gomote: swarming task creation using golangbuild name=%s", name)
cipdPlatform, ok := dimensions["cipd_platform"]
if !ok {
return "", fmt.Errorf("dimensions require cipd_platform: instance=%s", name)
}
goos, goarch, err := platformToGoValues(cipdPlatform)
if err != nil {
return "", err
}
packages := []*swarmpb.CipdPackage{
{Path: "tools/bin", PackageName: "infra/experimental/golangbuild/" + cipdPlatform, Version: "latest"},
{Path: "tools/bin", PackageName: "infra/tools/cipd/" + cipdPlatform, Version: "latest"},
{Path: "tools/bin", PackageName: "infra/tools/luci-auth/" + cipdPlatform, Version: "latest"},
}
pythonBin := "python3"
golangbuildBin := "golangbuild"
switch goos {
case "darwin":
golangbuildBin = `tools/bin/golangbuild`
pythonBin = `tools/bin/python3`
packages = append(packages,
&swarmpb.CipdPackage{Path: "tools", PackageName: "infra/3pp/tools/cpython3/" + cipdPlatform, Version: "latest"})
case "windows":
golangbuildBin = `tools\\bin\\golangbuild.exe`
pythonBin = `tools\bin\python3.exe`
packages = append(packages, &swarmpb.CipdPackage{Path: "tools", PackageName: "infra/3pp/tools/cpython3/" + cipdPlatform, Version: "latest"})
}
req := &swarmpb.NewTaskRequest{
Name: name,
Priority: 20, // 30 is the priority for builds
ServiceAccount: "coordinator-builder@golang-ci-luci.iam.gserviceaccount.com",
TaskSlices: []*swarmpb.TaskSlice{
&swarmpb.TaskSlice{
Properties: &swarmpb.TaskProperties{
CipdInput: &swarmpb.CipdInput{
Packages: packages,
},
EnvPrefixes: []*swarmpb.StringListPair{
{Key: "PATH", Value: []string{"tools/bin"}},
},
Command: []string{pythonBin, "-c", golangbuildStartup(goos, goarch, golangbuildBin)},
Dimensions: createStringPairs(dimensions),
Env: []*swarmpb.StringPair{
&swarmpb.StringPair{
Key: "GOMOTEID",
Value: name,
},
&swarmpb.StringPair{
Key: "GOMOTE_SETUP",
Value: properties.BuilderId,
},
},
ExecutionTimeoutSecs: 86400,
},
ExpirationSecs: 86400,
WaitForCapacity: false,
},
},
Tags: []string{"golang_mode:gomote"},
Realm: "golang:ci",
}
taskMD, err := ss.swarmingClient.NewTask(ctx, req)
if err != nil {
log.Printf("gomote: swarming task creation failed name=%s: %s", name, err)
return "", fmt.Errorf("unable to start task: %w", err)
}
log.Printf("gomote: task created: id=%s https://chromium-swarm.appspot.com/task?id=%s", taskMD.TaskId, taskMD.TaskId)
return taskMD.TaskId, nil
}
func condRun(fn func()) {
if fn != nil {
fn()
}
}
// tryThenPeriodicallyDo calls f and then calls f every period until the provided context is cancelled.
func tryThenPeriodicallyDo(ctx context.Context, period time.Duration, f func(context.Context, time.Time)) {
f(ctx, time.Now())
internal.PeriodicallyDo(ctx, period, f)
}
type configProperties struct {
BootstrapVersion string `json:"bootstrap_version"`
Mode int `json:"mode"`
BuilderId string // <project>/<bucket>/<builder>
}
func builderProperties(builder *buildbucketpb.BuilderItem) (*configProperties, error) {
cp := new(configProperties)
if err := json.Unmarshal([]byte(builder.GetConfig().GetProperties()), cp); err != nil {
return nil, fmt.Errorf("builder property unmarshal error: %s", err)
}
if cp.BootstrapVersion == "" {
cp.BootstrapVersion = "latest"
}
id := builder.GetId()
cp.BuilderId = fmt.Sprintf("%s/%s/%s", id.GetProject(), id.GetBucket(), id.GetBuilder())
return cp, nil
}