Files
opentf/internal/backend/remote-state/cos/client.go
Martin Atkins 67a5cd0911 statemgr+remote: context.Context parameters
This extends statemgr.Persistent, statemgr.Locker and remote.Client to
all expect context.Context parameters, and then updates all of the existing
implementations of those interfaces to support them.

All of the calls to statemgr.Persistent and statemgr.Locker methods outside
of tests are consistently context.TODO() for now, because the caller
landscape of these interfaces has some complications:

1. statemgr.Locker is also used by the clistate package for its state
   implementation that was derived from statemgr.Filesystem's predecessor,
   even though what clistate manages is not actually "state" in the sense
   of package statemgr. The callers of that are not yet ready to provide
   real contexts.

   In a future commit we'll either need to plumb context through to all of
   the clistate callers, or continue the effort to separate statemgr from
   clistate by introducing a clistate-specific "locker" API for it
   to use instead.

2. We call statemgr.Persistent and statemgr.Locker methods in situations
   where the active context might have already been cancelled, and so we'll
   need to make sure to ignore cancellation when calling those.

   This is mainly limited to PersistState and Unlock, since both need to
   be able to complete after a cancellation, but there are various
   codepaths that perform a Lock, Refresh, Persist, Unlock sequence and so
   it isn't yet clear where is the best place to enforce the invariant that
   Persist and Unlock must not be called with a cancelable context. We'll
   deal with that more in subsequent commits.

Within the various state manager and remote client implementations the
contexts _are_ wired together as best as possible with how these subsystems
are already laid out, and so once we deal with the problems above and make
callers provide suitable contexts they should be able to reach all of the
leaf API clients that might want to generate OpenTelemetry traces.

Signed-off-by: Martin Atkins <mart@degeneration.co.uk>
2025-07-10 08:11:39 -07:00

460 lines
11 KiB
Go

// Copyright (c) The OpenTofu Authors
// SPDX-License-Identifier: MPL-2.0
// Copyright (c) 2023 HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0
package cos
import (
"bytes"
"context"
"crypto/md5"
"encoding/json"
"errors"
"fmt"
"io"
"log"
"net/http"
"strings"
"time"
multierror "github.com/hashicorp/go-multierror"
"github.com/opentofu/opentofu/internal/states/remote"
"github.com/opentofu/opentofu/internal/states/statemgr"
tag "github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/tag/v20180813"
"github.com/tencentyun/cos-go-sdk-v5"
)
const (
lockTagKey = "tencentcloud-terraform-lock"
)
// RemoteClient implements the client of remote state
type remoteClient struct {
cosClient *cos.Client
tagClient *tag.Client
bucket string
stateFile string
lockFile string
encrypt bool
acl string
}
// Get returns remote state file
func (c *remoteClient) Get(ctx context.Context) (*remote.Payload, error) {
log.Printf("[DEBUG] get remote state file %s", c.stateFile)
exists, data, checksum, err := c.getObject(ctx, c.stateFile)
if err != nil {
return nil, err
}
if !exists {
return nil, nil
}
payload := &remote.Payload{
Data: data,
MD5: []byte(checksum),
}
return payload, nil
}
// Put put state file to remote
func (c *remoteClient) Put(ctx context.Context, data []byte) error {
log.Printf("[DEBUG] put remote state file %s", c.stateFile)
return c.putObject(ctx, c.stateFile, data)
}
// Delete delete remote state file
func (c *remoteClient) Delete(ctx context.Context) error {
log.Printf("[DEBUG] delete remote state file %s", c.stateFile)
return c.deleteObject(ctx, c.stateFile)
}
// Lock lock remote state file for writing
func (c *remoteClient) Lock(ctx context.Context, info *statemgr.LockInfo) (string, error) {
log.Printf("[DEBUG] lock remote state file %s", c.lockFile)
err := c.cosLock(c.bucket, c.lockFile)
if err != nil {
return "", c.lockError(ctx, err)
}
// Local helper function so we can call it multiple places
lockUnlock := func(parent error) error {
if err := c.cosUnlock(c.bucket, c.lockFile); err != nil {
return errors.Join(
fmt.Errorf("error unlocking cos state: %w", c.lockError(ctx, err)),
parent,
)
}
return parent
}
exists, _, _, err := c.getObject(ctx, c.lockFile)
if err != nil {
return "", lockUnlock(c.lockError(ctx, err))
}
if exists {
return "", lockUnlock(c.lockError(ctx, fmt.Errorf("lock file %s exists", c.lockFile)))
}
info.Path = c.lockFile
data, err := json.Marshal(info)
if err != nil {
return "", lockUnlock(c.lockError(ctx, err))
}
check := fmt.Sprintf("%x", md5.Sum(data))
err = c.putObject(ctx, c.lockFile, data)
if err != nil {
return "", lockUnlock(c.lockError(ctx, err))
}
return check, lockUnlock(nil)
}
// Unlock unlock remote state file
func (c *remoteClient) Unlock(ctx context.Context, check string) error {
log.Printf("[DEBUG] unlock remote state file %s", c.lockFile)
info, err := c.lockInfo(ctx)
if err != nil {
return c.lockError(ctx, err)
}
if info.ID != check {
return c.lockError(ctx, fmt.Errorf("lock id mismatch, %v != %v", info.ID, check))
}
err = c.deleteObject(ctx, c.lockFile)
if err != nil {
return c.lockError(ctx, err)
}
err = c.cosUnlock(c.bucket, c.lockFile)
if err != nil {
return c.lockError(ctx, err)
}
return nil
}
// lockError returns statemgr.LockError
func (c *remoteClient) lockError(ctx context.Context, err error) *statemgr.LockError {
log.Printf("[DEBUG] failed to lock or unlock %s: %v", c.lockFile, err)
lockErr := &statemgr.LockError{
Err: err,
}
info, infoErr := c.lockInfo(ctx)
if infoErr != nil {
lockErr.Err = multierror.Append(lockErr.Err, infoErr)
} else {
lockErr.Info = info
}
return lockErr
}
// lockInfo returns LockInfo from lock file
func (c *remoteClient) lockInfo(ctx context.Context) (*statemgr.LockInfo, error) {
exists, data, checksum, err := c.getObject(ctx, c.lockFile)
if err != nil {
return nil, err
}
if !exists {
return nil, fmt.Errorf("lock file %s not exists", c.lockFile)
}
info := &statemgr.LockInfo{}
if err := json.Unmarshal(data, info); err != nil {
return nil, err
}
info.ID = checksum
return info, nil
}
// getObject get remote object
func (c *remoteClient) getObject(ctx context.Context, cosFile string) (exists bool, data []byte, checksum string, err error) {
rsp, err := c.cosClient.Object.Get(ctx, cosFile, nil)
if rsp == nil {
log.Printf("[DEBUG] getObject %s: error: %v", cosFile, err)
err = fmt.Errorf("failed to open file at %v: %w", cosFile, err)
return
}
defer rsp.Body.Close()
log.Printf("[DEBUG] getObject %s: code: %d, error: %v", cosFile, rsp.StatusCode, err)
if err != nil {
if rsp.StatusCode == 404 {
err = nil
} else {
err = fmt.Errorf("failed to open file at %v: %w", cosFile, err)
}
return
}
checksum = rsp.Header.Get("X-Cos-Meta-Md5")
log.Printf("[DEBUG] getObject %s: checksum: %s", cosFile, checksum)
if len(checksum) != 32 {
err = fmt.Errorf("failed to open file at %v: checksum %s invalid", cosFile, checksum)
return
}
exists = true
data, err = io.ReadAll(rsp.Body)
log.Printf("[DEBUG] getObject %s: data length: %d", cosFile, len(data))
if err != nil {
err = fmt.Errorf("failed to open file at %v: %w", cosFile, err)
return
}
check := fmt.Sprintf("%x", md5.Sum(data))
log.Printf("[DEBUG] getObject %s: check: %s", cosFile, check)
if check != checksum {
err = fmt.Errorf("failed to open file at %v: checksum mismatch, %s != %s", cosFile, check, checksum)
return
}
return
}
// putObject put object to remote
func (c *remoteClient) putObject(ctx context.Context, cosFile string, data []byte) error {
opt := &cos.ObjectPutOptions{
ObjectPutHeaderOptions: &cos.ObjectPutHeaderOptions{
XCosMetaXXX: &http.Header{
"X-Cos-Meta-Md5": []string{fmt.Sprintf("%x", md5.Sum(data))},
},
},
ACLHeaderOptions: &cos.ACLHeaderOptions{
XCosACL: c.acl,
},
}
if c.encrypt {
opt.ObjectPutHeaderOptions.XCosServerSideEncryption = "AES256"
}
r := bytes.NewReader(data)
rsp, err := c.cosClient.Object.Put(ctx, cosFile, r, opt)
if rsp == nil {
log.Printf("[DEBUG] putObject %s: error: %v", cosFile, err)
return fmt.Errorf("failed to save file to %v: %w", cosFile, err)
}
defer rsp.Body.Close()
log.Printf("[DEBUG] putObject %s: code: %d, error: %v", cosFile, rsp.StatusCode, err)
if err != nil {
return fmt.Errorf("failed to save file to %v: %w", cosFile, err)
}
return nil
}
// deleteObject delete remote object
func (c *remoteClient) deleteObject(ctx context.Context, cosFile string) error {
rsp, err := c.cosClient.Object.Delete(ctx, cosFile)
if rsp == nil {
log.Printf("[DEBUG] deleteObject %s: error: %v", cosFile, err)
return fmt.Errorf("failed to delete file %v: %w", cosFile, err)
}
defer rsp.Body.Close()
log.Printf("[DEBUG] deleteObject %s: code: %d, error: %v", cosFile, rsp.StatusCode, err)
if rsp.StatusCode == 404 {
return nil
}
if err != nil {
return fmt.Errorf("failed to delete file %v: %w", cosFile, err)
}
return nil
}
// getBucket list bucket by prefix
func (c *remoteClient) getBucket(ctx context.Context, prefix string) (obs []cos.Object, err error) {
fs, rsp, err := c.cosClient.Bucket.Get(ctx, &cos.BucketGetOptions{Prefix: prefix})
if rsp == nil {
log.Printf("[DEBUG] getBucket %s/%s: error: %v", c.bucket, prefix, err)
err = fmt.Errorf("bucket %s not exists", c.bucket)
return
}
defer rsp.Body.Close()
log.Printf("[DEBUG] getBucket %s/%s: code: %d, error: %v", c.bucket, prefix, rsp.StatusCode, err)
if rsp.StatusCode == 404 {
err = fmt.Errorf("bucket %s not exists", c.bucket)
return
}
if err != nil {
return
}
return fs.Contents, nil
}
// putBucket create cos bucket
func (c *remoteClient) putBucket(ctx context.Context) error {
rsp, err := c.cosClient.Bucket.Put(ctx, nil)
if rsp == nil {
log.Printf("[DEBUG] putBucket %s: error: %v", c.bucket, err)
return fmt.Errorf("failed to create bucket %v: %w", c.bucket, err)
}
defer rsp.Body.Close()
log.Printf("[DEBUG] putBucket %s: code: %d, error: %v", c.bucket, rsp.StatusCode, err)
if rsp.StatusCode == 409 {
return nil
}
if err != nil {
return fmt.Errorf("failed to create bucket %v: %w", c.bucket, err)
}
return nil
}
// deleteBucket delete cos bucket
func (c *remoteClient) deleteBucket(ctx context.Context, recursive bool) error {
if recursive {
obs, err := c.getBucket(ctx, "")
if err != nil {
if strings.Contains(err.Error(), "not exists") {
return nil
}
log.Printf("[DEBUG] deleteBucket %s: empty bucket error: %v", c.bucket, err)
return fmt.Errorf("failed to empty bucket %v: %w", c.bucket, err)
}
for _, v := range obs {
err := c.deleteObject(ctx, v.Key)
if err != nil {
return fmt.Errorf("failed to delete object with key %s: %w", v.Key, err)
}
}
}
rsp, err := c.cosClient.Bucket.Delete(ctx)
if rsp == nil {
log.Printf("[DEBUG] deleteBucket %s: error: %v", c.bucket, err)
return fmt.Errorf("failed to delete bucket %v: %w", c.bucket, err)
}
defer rsp.Body.Close()
log.Printf("[DEBUG] deleteBucket %s: code: %d, error: %v", c.bucket, rsp.StatusCode, err)
if rsp.StatusCode == 404 {
return nil
}
if err != nil {
return fmt.Errorf("failed to delete bucket %v: %w", c.bucket, err)
}
return nil
}
// cosLock lock cos for writing
func (c *remoteClient) cosLock(bucket, cosFile string) error {
log.Printf("[DEBUG] lock cos file %s:%s", bucket, cosFile)
cosPath := fmt.Sprintf("%s:%s", bucket, cosFile)
lockTagValue := fmt.Sprintf("%x", md5.Sum([]byte(cosPath)))
return c.CreateTag(lockTagKey, lockTagValue)
}
// cosUnlock unlock cos writing
func (c *remoteClient) cosUnlock(bucket, cosFile string) error {
log.Printf("[DEBUG] unlock cos file %s:%s", bucket, cosFile)
cosPath := fmt.Sprintf("%s:%s", bucket, cosFile)
lockTagValue := fmt.Sprintf("%x", md5.Sum([]byte(cosPath)))
var err error
for i := 0; i < 30; i++ {
tagExists, err := c.CheckTag(lockTagKey, lockTagValue)
if err != nil {
return err
}
if !tagExists {
return nil
}
err = c.DeleteTag(lockTagKey, lockTagValue)
if err == nil {
return nil
}
time.Sleep(1 * time.Second)
}
return err
}
// CheckTag checks if tag key:value exists
func (c *remoteClient) CheckTag(key, value string) (exists bool, err error) {
request := tag.NewDescribeTagsRequest()
request.TagKey = &key
request.TagValue = &value
response, err := c.tagClient.DescribeTags(request)
log.Printf("[DEBUG] create tag %s:%s: error: %v", key, value, err)
if err != nil {
return
}
if len(response.Response.Tags) == 0 {
return
}
tagKey := response.Response.Tags[0].TagKey
tagValue := response.Response.Tags[0].TagValue
exists = key == *tagKey && value == *tagValue
return
}
// CreateTag create tag by key and value
func (c *remoteClient) CreateTag(key, value string) error {
request := tag.NewCreateTagRequest()
request.TagKey = &key
request.TagValue = &value
_, err := c.tagClient.CreateTag(request)
log.Printf("[DEBUG] create tag %s:%s: error: %v", key, value, err)
if err != nil {
return fmt.Errorf("failed to create tag: %s -> %s: %w", key, value, err)
}
return nil
}
// DeleteTag create tag by key and value
func (c *remoteClient) DeleteTag(key, value string) error {
request := tag.NewDeleteTagRequest()
request.TagKey = &key
request.TagValue = &value
_, err := c.tagClient.DeleteTag(request)
log.Printf("[DEBUG] delete tag %s:%s: error: %v", key, value, err)
if err != nil {
return fmt.Errorf("failed to delete tag: %s -> %s: %w", key, value, err)
}
return nil
}