How to implement cross-region replication for a bucket using Go SDK V2 - Object Storage Service

OSS data replication automatically synchronizes objects — including create, overwrite, and delete operations — from a source bucket to a destination bucket. Two modes are available: cross-region replication (CRR) replicates data across regions, and same-region replication (SRR) replicates data within the same region.

Prerequisites

Before you begin, make sure you have:

The OSS Go SDK V2 installed (github.com/aliyun/alibabacloud-oss-go-sdk-v2)
Access credentials stored in environment variables. For setup instructions, see Configure access credentials.

The required permissions. Alibaba Cloud accounts have replication permissions by default. RAM users and Security Token Service (STS) temporary credentials require explicit grants:

Operation	Required permission
Enable data replication	`oss:PutBucketReplication`
Enable or disable RTC	`oss:PutBucketRtc`
View replication rules	`oss:GetBucketReplication`
View destination regions	`oss:GetBucketReplicationLocation`
View replication task progress	`oss:GetBucketReplicationProgress`
Disable data replication	`oss:DeleteBucketReplication`

The examples in this topic use the China (Hangzhou) region (cn-hangzhou) with a public endpoint. To access OSS from other Alibaba Cloud services in the same region, use an internal endpoint instead. For supported regions and endpoints, see OSS regions and endpoints.

Enable data replication

Important

Before enabling data replication, make sure both the source and destination buckets are either unversioned or versioning-enabled.

The following example shows how to enable data replication. The source and destination buckets may be located in the same region or different regions. This example uses a destination bucket in oss-cn-beijing, with replication time control (RTC) and historical object replication both enabled.

Key parameters in PutBucketReplicationRequest:

Parameter	Description
`Bucket`	Source bucket name
`Destination.Bucket`	Destination bucket name
`Destination.Location`	Region of the destination bucket (e.g., `oss-cn-beijing`)
`Destination.TransferType`	Transfer method. Use `oss.TransferTypeOssAcc` for OSS transfer acceleration
`RTC.Status`	Set to `"enabled"` to enable RTC
`HistoricalObjectReplication`	Set to `oss.HistoricalObjectReplicationEnabled` to replicate existing objects

package main

import (
	"context"
	"flag"
	"log"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
)

var (
	region     string
	bucketName string
)

func init() {
	flag.StringVar(&region, "region", "", "The region in which the bucket is located.")
	flag.StringVar(&bucketName, "bucket", "", "The name of the bucket.")
}

func main() {
	flag.Parse()

	var (
		targetBucket   = "target bucket name"
		targetLocation = "oss-cn-beijing"
	)

	if len(bucketName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, bucket name required")
	}

	if len(region) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, region required")
	}

	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	client := oss.NewClient(cfg)

	request := &oss.PutBucketReplicationRequest{
		Bucket: oss.Ptr(bucketName),
		ReplicationConfiguration: &oss.ReplicationConfiguration{
			Rules: []oss.ReplicationRule{
				{
					RTC: &oss.ReplicationTimeControl{
						Status: oss.Ptr("enabled"), // Enable RTC.
					},
					Destination: &oss.ReplicationDestination{
						Bucket:       oss.Ptr(targetBucket),
						Location:     oss.Ptr(targetLocation),
						TransferType: oss.TransferTypeOssAcc,
					},
					HistoricalObjectReplication: oss.HistoricalObjectReplicationEnabled,
				},
			},
		},
	}

	result, err := client.PutBucketReplication(context.TODO(), request)
	if err != nil {
		log.Fatalf("failed to put bucket replication %v", err)
	}

	log.Printf("put bucket replication result:%#v\n", result)
}

Query data replication rules

Retrieve all replication rules configured for a bucket.

package main

import (
	"context"
	"flag"
	"log"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
)

var (
	region     string
	bucketName string
)

func init() {
	flag.StringVar(&region, "region", "", "The region in which the bucket is located.")
	flag.StringVar(&bucketName, "bucket", "", "The name of the bucket.")
}

func main() {
	flag.Parse()

	if len(bucketName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, bucket name required")
	}

	if len(region) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, region required")
	}

	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	client := oss.NewClient(cfg)

	request := &oss.GetBucketReplicationRequest{
		Bucket: oss.Ptr(bucketName),
	}

	result, err := client.GetBucketReplication(context.TODO(), request)
	if err != nil {
		log.Fatalf("failed to get bucket replication %v", err)
	}

	log.Printf("get bucket replication result:%#v\n", result)
}

Enable or disable the RTC feature

Update the replication time control (RTC) setting for an existing CRR rule.

Key parameters in PutBucketRtcRequest:

Parameter	Description
`Bucket`	Source bucket name
`RtcConfiguration.ID`	ID of the existing replication rule to update
`RtcConfiguration.RTC.Status`	Set to `"enabled"` to enable RTC, or `"disabled"` to disable it

package main

import (
	"context"
	"flag"
	"log"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
)

var (
	region     string
	bucketName string
)

func init() {
	flag.StringVar(&region, "region", "", "The region in which the bucket is located.")
	flag.StringVar(&bucketName, "bucket", "", "The name of the bucket.")
}

func main() {
	flag.Parse()

	var ruleId = "replication id"

	if len(bucketName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, bucket name required")
	}

	if len(region) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, region required")
	}

	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	client := oss.NewClient(cfg)

	request := &oss.PutBucketRtcRequest{
		Bucket: oss.Ptr(bucketName),
		RtcConfiguration: &oss.RtcConfiguration{
			RTC: &oss.ReplicationTimeControl{
				Status: oss.Ptr("enabled"),
			},
			ID: oss.Ptr(ruleId), // ID of the replication rule to update.
		},
	}

	result, err := client.PutBucketRtc(context.TODO(), request)
	if err != nil {
		log.Fatalf("failed to put bucket rtc %v", err)
	}

	log.Printf("put bucket rtc result:%#v\n", result)
}

Query destination regions

List the regions to which data can be replicated from a bucket.

package main

import (
	"context"
	"flag"
	"log"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
)

var (
	region     string
	bucketName string
)

func init() {
	flag.StringVar(&region, "region", "", "The region in which the bucket is located.")
	flag.StringVar(&bucketName, "bucket", "", "The name of the bucket.")
}

func main() {
	flag.Parse()

	if len(bucketName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, bucket name required")
	}

	if len(region) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, region required")
	}

	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	client := oss.NewClient(cfg)

	request := &oss.GetBucketReplicationLocationRequest{
		Bucket: oss.Ptr(bucketName),
	}

	result, err := client.GetBucketReplicationLocation(context.TODO(), request)
	if err != nil {
		log.Fatalf("failed to get bucket replication location %v", err)
	}

	for _, location := range result.ReplicationLocation.Locations {
		log.Printf("Bucket replication location: %s", location)
	}

	for _, rtcLocation := range result.ReplicationLocation.LocationRTCConstraint.Locations {
		log.Printf("Bucket replication location (RTC supported): %s", rtcLocation)
	}
}

Query replication task progress

OSS tracks two types of replication progress:

Historical data replication: expressed as a percentage. Available only when historical object replication is enabled for the rule.
Incremental data replication: expressed as a point in time. Objects stored in the source bucket before this timestamp have been replicated.

This example queries progress for a replication rule identified by its rule ID.

package main

import (
	"context"
	"flag"
	"log"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
)

var (
	region     string
	bucketName string
)

func init() {
	flag.StringVar(&region, "region", "", "The region in which the bucket is located.")
	flag.StringVar(&bucketName, "bucket", "", "The name of the bucket.")
}

func main() {
	flag.Parse()

	var ruleId = "replication id"

	if len(bucketName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, bucket name required")
	}

	if len(region) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, region required")
	}

	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	client := oss.NewClient(cfg)

	request := &oss.GetBucketReplicationProgressRequest{
		Bucket: oss.Ptr(bucketName),
		RuleId: oss.Ptr(ruleId),
	}

	result, err := client.GetBucketReplicationProgress(context.TODO(), request)
	if err != nil {
		log.Fatalf("failed to get bucket replication progress %v", err)
	}

	for _, rule := range result.ReplicationProgress.Rules {
		log.Printf("Rule ID: %s", rule.ID)
		log.Printf("Status: %s", rule.Status)
		log.Printf("Destination bucket: %s", *rule.Destination.Bucket)
		log.Printf("Destination location: %s", *rule.Destination.Location)
		log.Printf("Transfer type: %v", rule.Destination.TransferType)
		log.Printf("Historical object replication: %s", *rule.HistoricalObjectReplication)
	}
}

Disable data replication

Delete a replication rule by its rule ID to stop replication for that rule.

package main

import (
	"context"
	"flag"
	"log"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
)

var (
	region     string
	bucketName string
)

func init() {
	flag.StringVar(&region, "region", "", "The region in which the bucket is located.")
	flag.StringVar(&bucketName, "bucket", "", "The name of the bucket.")
}

func main() {
	flag.Parse()

	var ruleId = "replication id"

	if len(bucketName) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, bucket name required")
	}

	if len(region) == 0 {
		flag.PrintDefaults()
		log.Fatalf("invalid parameters, region required")
	}

	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	client := oss.NewClient(cfg)

	request := &oss.DeleteBucketReplicationRequest{
		Bucket: oss.Ptr(bucketName),
		ReplicationRules: &oss.ReplicationRules{
			IDs: []string{ruleId},
		},
	}

	result, err := client.DeleteBucketReplication(context.TODO(), request)
	if err != nil {
		log.Fatalf("failed to delete bucket replication %v", err)
	}

	log.Printf("delete bucket replication result:%#v\n", result)
}

API reference

Operation	API reference
Enable data replication	PutBucketReplication
Enable or disable RTC	PutBucketRTC
Query replication rules	GetBucketReplication
Query destination regions	GetBucketReplicationLocation
Query replication task progress	GetBucketReplicationProgress
Disable data replication	DeleteBucketReplication