refactor: move source implementation in Invoke() function to Source

internal/sources/dataplex/dataplex.go

@@ -19,6 +19,8 @@ import (
 	"fmt"
 
 	dataplexapi "cloud.google.com/go/dataplex/apiv1"
+	"cloud.google.com/go/dataplex/apiv1/dataplexpb"
+	"github.com/cenkalti/backoff/v5"
 	"github.com/goccy/go-yaml"
 	"github.com/googleapis/genai-toolbox/internal/sources"
 	"github.com/googleapis/genai-toolbox/internal/util"
@@ -121,3 +123,101 @@ func initDataplexConnection(
 	}
 	return client, nil
 }
+
+func (s *Source) LookupEntry(ctx context.Context, name string, view int, aspectTypes []string, entry string) (*dataplexpb.Entry, error) {
+	viewMap := map[int]dataplexpb.EntryView{
+		1: dataplexpb.EntryView_BASIC,
+		2: dataplexpb.EntryView_FULL,
+		3: dataplexpb.EntryView_CUSTOM,
+		4: dataplexpb.EntryView_ALL,
+	}
+	req := &dataplexpb.LookupEntryRequest{
+		Name:        name,
+		View:        viewMap[view],
+		AspectTypes: aspectTypes,
+		Entry:       entry,
+	}
+	result, err := s.CatalogClient().LookupEntry(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+	return result, nil
+}
+
+func (s *Source) searchRequest(ctx context.Context, query string, pageSize int, orderBy string) (*dataplexapi.SearchEntriesResultIterator, error) {
+	// Create SearchEntriesRequest with the provided parameters
+	req := &dataplexpb.SearchEntriesRequest{
+		Query:          query,
+		Name:           fmt.Sprintf("projects/%s/locations/global", s.ProjectID()),
+		PageSize:       int32(pageSize),
+		OrderBy:        orderBy,
+		SemanticSearch: true,
+	}
+
+	// Perform the search using the CatalogClient - this will return an iterator
+	it := s.CatalogClient().SearchEntries(ctx, req)
+	if it == nil {
+		return nil, fmt.Errorf("failed to create search entries iterator for project %q", s.ProjectID())
+	}
+	return it, nil
+}
+
+func (s *Source) SearchAspectTypes(ctx context.Context, query string, pageSize int, orderBy string) ([]*dataplexpb.AspectType, error) {
+	q := query + " type=projects/dataplex-types/locations/global/entryTypes/aspecttype"
+	it, err := s.searchRequest(ctx, q, pageSize, orderBy)
+	if err != nil {
+		return nil, err
+	}
+
+	// Iterate through the search results and call GetAspectType for each result using the resource name
+	var results []*dataplexpb.AspectType
+	for {
+		entry, err := it.Next()
+		if err != nil {
+			break
+		}
+
+		// Create an instance of exponential backoff with default values for retrying GetAspectType calls
+		// InitialInterval, RandomizationFactor, Multiplier, MaxInterval = 500 ms, 0.5, 1.5, 60 s
+		getAspectBackOff := backoff.NewExponentialBackOff()
+
+		resourceName := entry.DataplexEntry.GetEntrySource().Resource
+		getAspectTypeReq := &dataplexpb.GetAspectTypeRequest{
+			Name: resourceName,
+		}
+
+		operation := func() (*dataplexpb.AspectType, error) {
+			aspectType, err := s.CatalogClient().GetAspectType(ctx, getAspectTypeReq)
+			if err != nil {
+				return nil, fmt.Errorf("failed to get aspect type for entry %q: %w", resourceName, err)
+			}
+			return aspectType, nil
+		}
+
+		// Retry the GetAspectType operation with exponential backoff
+		aspectType, err := backoff.Retry(ctx, operation, backoff.WithBackOff(getAspectBackOff))
+		if err != nil {
+			return nil, fmt.Errorf("failed to get aspect type after retries for entry %q: %w", resourceName, err)
+		}
+
+		results = append(results, aspectType)
+	}
+	return results, nil
+}
+
+func (s *Source) SearchEntries(ctx context.Context, query string, pageSize int, orderBy string) ([]*dataplexpb.SearchEntriesResult, error) {
+	it, err := s.searchRequest(ctx, query, pageSize, orderBy)
+	if err != nil {
+		return nil, err
+	}
+
+	var results []*dataplexpb.SearchEntriesResult
+	for {
+		entry, err := it.Next()
+		if err != nil {
+			break
+		}
+		results = append(results, entry)
+	}
+	return results, nil
+}

internal/sources/http/http.go

@@ -16,7 +16,9 @@ package http
 import (
 	"context"
 	"crypto/tls"
+	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
 	"net/url"
 	"time"
@@ -143,3 +145,28 @@ func (s *Source) HttpQueryParams() map[string]string {
 func (s *Source) Client() *http.Client {
 	return s.client
 }
+
+func (s *Source) RunRequest(req *http.Request) (any, error) {
+	// Make request and fetch response
+	resp, err := s.Client().Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("error making HTTP request: %s", err)
+	}
+	defer resp.Body.Close()
+
+	var body []byte
+	body, err = io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode < 200 || resp.StatusCode > 299 {
+		return nil, fmt.Errorf("unexpected status code: %d, response body: %s", resp.StatusCode, string(body))
+	}
+
+	var data any
+	if err = json.Unmarshal(body, &data); err != nil {
+		// if unable to unmarshal data, return result as string.
+		return string(body), nil
+	}
+	return data, nil
+}

internal/sources/serverlessspark/serverlessspark.go

@@ -16,15 +16,21 @@ package serverlessspark
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
+	"time"
 
 	dataproc "cloud.google.com/go/dataproc/v2/apiv1"
+	"cloud.google.com/go/dataproc/v2/apiv1/dataprocpb"
 	longrunning "cloud.google.com/go/longrunning/autogen"
+	"cloud.google.com/go/longrunning/autogen/longrunningpb"
 	"github.com/goccy/go-yaml"
 	"github.com/googleapis/genai-toolbox/internal/sources"
 	"github.com/googleapis/genai-toolbox/internal/util"
 	"go.opentelemetry.io/otel/trace"
+	"google.golang.org/api/iterator"
 	"google.golang.org/api/option"
+	"google.golang.org/protobuf/encoding/protojson"
 )
 
 const SourceKind string = "serverless-spark"
@@ -121,3 +127,168 @@ func (s *Source) Close() error {
 	}
 	return nil
 }
+
+func (s *Source) CancelOperation(ctx context.Context, operation string) (any, error) {
+	req := &longrunningpb.CancelOperationRequest{
+		Name: fmt.Sprintf("projects/%s/locations/%s/operations/%s", s.GetProject(), s.GetLocation(), operation),
+	}
+	client, err := s.GetOperationsClient(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get operations client: %w", err)
+	}
+	err = client.CancelOperation(ctx, req)
+	if err != nil {
+		return nil, fmt.Errorf("failed to cancel operation: %w", err)
+	}
+	return fmt.Sprintf("Cancelled [%s].", operation), nil
+}
+
+func (s *Source) CreateBatch(ctx context.Context, batch *dataprocpb.Batch) (map[string]any, error) {
+	req := &dataprocpb.CreateBatchRequest{
+		Parent: fmt.Sprintf("projects/%s/locations/%s", s.GetProject(), s.GetLocation()),
+		Batch:  batch,
+	}
+
+	client := s.GetBatchControllerClient()
+	op, err := client.CreateBatch(ctx, req)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create batch: %w", err)
+	}
+	meta, err := op.Metadata()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get create batch op metadata: %w", err)
+	}
+
+	projectID, location, batchID, err := ExtractBatchDetails(meta.GetBatch())
+	if err != nil {
+		return nil, fmt.Errorf("error extracting batch details from name %q: %v", meta.GetBatch(), err)
+	}
+	consoleUrl := BatchConsoleURL(projectID, location, batchID)
+	logsUrl := BatchLogsURL(projectID, location, batchID, meta.GetCreateTime().AsTime(), time.Time{})
+
+	wrappedResult := map[string]any{
+		"opMetadata": meta,
+		"consoleUrl": consoleUrl,
+		"logsUrl":    logsUrl,
+	}
+	return wrappedResult, nil
+}
+
+// ListBatchesResponse is the response from the list batches API.
+type ListBatchesResponse struct {
+	Batches       []Batch `json:"batches"`
+	NextPageToken string  `json:"nextPageToken"`
+}
+
+// Batch represents a single batch job.
+type Batch struct {
+	Name       string `json:"name"`
+	UUID       string `json:"uuid"`
+	State      string `json:"state"`
+	Creator    string `json:"creator"`
+	CreateTime string `json:"createTime"`
+	Operation  string `json:"operation"`
+	ConsoleURL string `json:"consoleUrl"`
+	LogsURL    string `json:"logsUrl"`
+}
+
+func (s *Source) ListBatches(ctx context.Context, ps *int, pt, filter string) (any, error) {
+	client := s.GetBatchControllerClient()
+	parent := fmt.Sprintf("projects/%s/locations/%s", s.GetProject(), s.GetLocation())
+	req := &dataprocpb.ListBatchesRequest{
+		Parent:  parent,
+		OrderBy: "create_time desc",
+	}
+
+	if ps != nil {
+		req.PageSize = int32(*ps)
+	}
+	if pt != "" {
+		req.PageToken = pt
+	}
+	if filter != "" {
+		req.Filter = filter
+	}
+
+	it := client.ListBatches(ctx, req)
+	pager := iterator.NewPager(it, int(req.PageSize), req.PageToken)
+
+	var batchPbs []*dataprocpb.Batch
+	nextPageToken, err := pager.NextPage(&batchPbs)
+	if err != nil {
+		return nil, fmt.Errorf("failed to list batches: %w", err)
+	}
+
+	batches, err := ToBatches(batchPbs)
+	if err != nil {
+		return nil, err
+	}
+
+	return ListBatchesResponse{Batches: batches, NextPageToken: nextPageToken}, nil
+}
+
+// ToBatches converts a slice of protobuf Batch messages to a slice of Batch structs.
+func ToBatches(batchPbs []*dataprocpb.Batch) ([]Batch, error) {
+	batches := make([]Batch, 0, len(batchPbs))
+	for _, batchPb := range batchPbs {
+		consoleUrl, err := BatchConsoleURLFromProto(batchPb)
+		if err != nil {
+			return nil, fmt.Errorf("error generating console url: %v", err)
+		}
+		logsUrl, err := BatchLogsURLFromProto(batchPb)
+		if err != nil {
+			return nil, fmt.Errorf("error generating logs url: %v", err)
+		}
+		batch := Batch{
+			Name:       batchPb.Name,
+			UUID:       batchPb.Uuid,
+			State:      batchPb.State.Enum().String(),
+			Creator:    batchPb.Creator,
+			CreateTime: batchPb.CreateTime.AsTime().Format(time.RFC3339),
+			Operation:  batchPb.Operation,
+			ConsoleURL: consoleUrl,
+			LogsURL:    logsUrl,
+		}
+		batches = append(batches, batch)
+	}
+	return batches, nil
+}
+
+func (s *Source) GetBatch(ctx context.Context, name string) (map[string]any, error) {
+	client := s.GetBatchControllerClient()
+	req := &dataprocpb.GetBatchRequest{
+		Name: fmt.Sprintf("projects/%s/locations/%s/batches/%s", s.GetProject(), s.GetLocation(), name),
+	}
+
+	batchPb, err := client.GetBatch(ctx, req)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get batch: %w", err)
+	}
+
+	jsonBytes, err := protojson.Marshal(batchPb)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal batch to JSON: %w", err)
+	}
+
+	var result map[string]any
+	if err := json.Unmarshal(jsonBytes, &result); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal batch JSON: %w", err)
+	}
+
+	consoleUrl, err := BatchConsoleURLFromProto(batchPb)
+	if err != nil {
+		return nil, fmt.Errorf("error generating console url: %v", err)
+	}
+	logsUrl, err := BatchLogsURLFromProto(batchPb)
+	if err != nil {
+		return nil, fmt.Errorf("error generating logs url: %v", err)
+	}
+
+	wrappedResult := map[string]any{
+		"consoleUrl": consoleUrl,
+		"logsUrl":    logsUrl,
+		"batch":      result,
+	}
+
+	return wrappedResult, nil
+}