【Cloud SQL】GolangでDBインスタンスへのオペレーション完了まで待つ

1
package mypkg
2

3
import (
4
  "context"
5
  "encoding/json"
6
  "fmt"
7
  "log"
8
  "time"
9

10
  "golang.org/x/oauth2/google"
11
  "google.golang.org/api/option"
12
  sqladmin "google.golang.org/api/sqladmin/v1beta4"
13

14
  _ "github.com/GoogleCloudPlatform/functions-framework-go/funcframework"
15
)
16

17
type PubSubMessage struct {
18
  Data []byte `json:"data"`
19
}
20

21
type MessagePayload struct {
22
  Instance          string
23
  Project           string
24
  Action            string
25
  DiscordWebhookUrl string
26
  CloudSqlIconUrl   string
27
}
28

29
func WaitOperation(ctx context.Context, svc *sqladmin.Service, prj string, instance string) error {
30
  // Use a backoff strategy for retries
31
  retryInterval := 10 * time.Second
32
  maxRetries := 60
33

34
  for i := 0; i < maxRetries; i++ {
35
    // See more details at: https://pkg.go.dev/google.golang.org/api/sqladmin/v1beta4#OperationsListCall
36
    ops, err := svc.Operations.List(prj).Instance(instance).Context(ctx).Do()
37
    if err != nil {
38
      return fmt.Errorf("failed to list operations: %w", err)
39
    }
40

41
    if len(ops.Items) > 0 {
42
      op := ops.Items[0] // Assuming we're interested in the most recent operation for the instance
43

44
      switch op.Status {
45
      case "DONE":
46
        log.Printf("Operation %s completed with status: %s", op.Name, op.Status)
47
        return nil // Operation completed successfully
48
      case "RUNNING":
49
        // Wait before retrying
50
        log.Printf("Operation %s is still running. Retrying in %v...", op.Name, retryInterval)
51
        time.Sleep(retryInterval)
52
        continue
53
      default: // Includes "FAILED" and other possible states
54
        return fmt.Errorf("operation %s failed with status: %s, Error: %v", op.Name, op.Status, op.Error)
55
      }
56
    } else {
57
      log.Printf("No operations found for instance %s. Retrying in %v...", instance, retryInterval)
58
      // time.Sleep(retryInterval)
59
      return nil // Operation completed successfully
60
    }
61
  }
62

63
  return fmt.Errorf("operation timed out after %v", time.Duration(maxRetries)*retryInterval)
64
}
65

66
// ProcessPubSub consumes and processes a Pub/Sub message.
67
func ProcessPubSub(ctx context.Context, m PubSubMessage) error {
68
  var psData MessagePayload
69
  log.Printf("Receiving a request with the following payload: %+v\n", psData)
70

71
  err := json.Unmarshal(m.Data, &psData)
72
  if err != nil {
73
    log.Println(err)
74
  }
75
  log.Printf("Received a request with the following payload: %+v\n", psData)
76
  log.Printf("Request received for Cloud SQL instance %s action: %s, %s", psData.Action, psData.Instance, psData.Project)
77

78
  // Create an http.Client that uses Application Default Credentials.
79
  hc, err := google.DefaultClient(ctx, sqladmin.CloudPlatformScope)
80
  if err != nil {
81
    return err
82
  }
83

84
  // Create the Google Cloud SQL service.
85
  service, err := sqladmin.NewService(ctx, option.WithHTTPClient(hc))
86
  if err != nil {
87
    return err
88
  }
89

90
  // Wait until running operations are completed.
91
  err = WaitOperation(ctx, service, psData.Project, psData.Instance)
92
  if err != nil {
93
    fmt.Printf("Error: %+v\n", err)
94
  }
95

96
  // my DB operations...
97

98
  return nil
99
}