Use Go SDK to implement cross-region replication for buckets-Object Storage Service(OSS)-阿里云帮助中心

Data replication automatically and asynchronously copies objects and their operations—creation, overwriting, and deletion—from a source bucket to a destination bucket in near real time. Object Storage Service (OSS) supports cross-region replication (CRR) and same-region replication (SRR).

When to use CRR, SRR, and RTC

Requirement	CRR	SRR	RTC
Replicate across different regions	Yes	No	—
Replicate within the same region	No	Yes	—
Guarantee replication within a predictable time window	No	No	Yes
Replicate historical objects created before the rule	Yes (optional)	Yes (optional)	—
Replicate SSE-KMS encrypted objects	Yes	Yes	—

Replication time control (RTC) applies to CRR rules only. Enable it when your workload requires a predictable replication time SLA.

Prerequisites

Before you begin, make sure that you have:

The Go SDK installed: github.com/aliyun/aliyun-oss-go-sdk/oss
Access credentials stored in the OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET environment variables. For setup instructions, see Configure access credentials
The required permissions for the operations you intend to perform

By default, an Alibaba Cloud account has permissions for all data replication operations. If you use a Resource Access Management (RAM) user or temporary access credentials from Security Token Service (STS), grant the permissions listed below.

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

Usage notes

The examples in this topic use the public endpoint of the China (Hangzhou) region. To access OSS from other Alibaba Cloud services in the same region, use an internal endpoint. For region and endpoint details, see Regions and endpoints.
The examples create an OSSClient instance using an OSS endpoint. To create an OSSClient instance with a custom domain name or STS, see Configure OSSClient instances.

Enable data replication

Before enabling replication, make sure that versioning is either disabled for both the source and destination buckets, or enabled for both.

PutBucketReplication accepts a replication rule with the following key parameters:

Parameter	Description
`PrefixSet`	Limits replication to objects matching the specified prefixes. Omit to replicate all objects.
`Action`	Operations to replicate. `PUT` replicates object creation and updates.
`Destination`	Specifies the destination bucket, region (`Location`), and transfer type (`TransferType`). Use `oss_acc` for accelerated cross-region transfer.
`HistoricalObjectReplication`	Set to `enabled` to replicate objects created before the rule takes effect. Set to `disabled` to replicate only new objects.
`RTC`	Set to `enabled` to turn on replication time control. Applies to CRR rules only.
`SourceSelectionCriteria`	Set to `Enabled` to replicate objects encrypted with SSE-KMS. Requires the `EncryptionConfiguration` (KMS key ID).
`SyncRole`	The RAM role that OSS assumes to replicate objects.

The following example replicates objects with prefixes prefix_1 and prefix_2 from srcexamplebucket in China (Hangzhou) to destexamplebucket, with RTC enabled and SSE-KMS object replication configured.

package main

import (
	"encoding/xml"
	"log"

	"github.com/aliyun/aliyun-oss-go-sdk/oss"
)

func main() {
	// Obtain access credentials from environment variables. Before you run this code,
	// make sure that the OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET environment variables are set.
	provider, err := oss.NewEnvironmentVariableCredentialsProvider()
	if err != nil {
		log.Fatalf("Failed to create credentials provider: %v", err)
	}

	// Create an OSSClient instance.
	// Set yourEndpoint to the endpoint of the source bucket.
	// Example: https://oss-cn-hangzhou.aliyuncs.com for China (Hangzhou).
	// Set yourRegion to the corresponding region ID, e.g., cn-hangzhou.
	clientOptions := []oss.ClientOption{oss.SetCredentialsProvider(&provider)}
	clientOptions = append(clientOptions, oss.Region("yourRegion"))
	clientOptions = append(clientOptions, oss.AuthVersion(oss.AuthV4))
	client, err := oss.New("yourEndpoint", "", "", clientOptions...)
	if err != nil {
		log.Fatalf("Failed to create OSS client: %v", err)
	}

	srcbucketName := "srcexamplebucket"
	destBucketName := "destexamplebucket"

	// Replicate only objects whose keys start with prefix_1 or prefix_2.
	// Remove PrefixSet from the rule to replicate all objects.
	prefix1 := "prefix_1"
	prefix2 := "prefix_2"
	prefixSet := oss.ReplicationRulePrefix{Prefix: []*string{&prefix1, &prefix2}}

	// Enable RTC for this CRR rule.
	enabled := "enabled"

	// KMS key ID used for SSE-KMS encrypted object replication.
	keyId := "c4d49f85-ee30-426b-a5ed-95e9139d"

	// Replicate SSE-KMS encrypted objects.
	source := "Enabled"

	reqReplication := oss.PutBucketReplication{
		Rule: []oss.ReplicationRule{
			{
				PrefixSet:                   &prefixSet,
				Action:                      "PUT", // Replicate object creation and updates.
				RTC:                         &enabled,
				Destination: &oss.ReplicationRuleDestination{
					Bucket:       destBucketName,
					Location:     "oss-cn-hangzhou",
					TransferType: "oss_acc", // Use accelerated transfer.
				},
				HistoricalObjectReplication: "disabled", // Replicate new objects only.
				SyncRole:                    "aliyunramrole",
				EncryptionConfiguration:     &keyId,
				SourceSelectionCriteria:     &source,
			},
		},
	}

	xmlBody, err := xml.Marshal(reqReplication)
	if err != nil {
		log.Fatalf("Failed to marshal XML for PutBucketReplication: %v", err)
	}
	err = client.PutBucketReplication(srcbucketName, string(xmlBody))
	if err != nil {
		log.Fatalf("Failed to put bucket replication: %v", err)
	}

	log.Println("Put Bucket Replication Success!")
}

Query replication rules

GetBucketReplication returns all replication rules for a bucket, including their IDs, destinations, status, and RTC settings. The rule ID returned here is required for RTC updates, progress queries, and rule deletion.

package main

import (
	"encoding/xml"
	"log"

	"github.com/aliyun/aliyun-oss-go-sdk/oss"
)

func main() {
	provider, err := oss.NewEnvironmentVariableCredentialsProvider()
	if err != nil {
		log.Fatalf("Failed to create credentials provider: %v", err)
	}

	clientOptions := []oss.ClientOption{oss.SetCredentialsProvider(&provider)}
	clientOptions = append(clientOptions, oss.Region("yourRegion"))
	clientOptions = append(clientOptions, oss.AuthVersion(oss.AuthV4))
	client, err := oss.New("yourEndpoint", "", "", clientOptions...)
	if err != nil {
		log.Fatalf("Failed to create OSS client: %v", err)
	}

	bucketName := "srcexamplebucket"

	stringData, err := client.GetBucketReplication(bucketName)
	if err != nil {
		log.Fatalf("Failed to get bucket replication: %v", err)
	}

	var repResult oss.GetBucketReplicationResult
	err = xml.Unmarshal([]byte(stringData), &repResult)
	if err != nil {
		log.Fatalf("Failed to unmarshal XML response: %v", err)
	}

	for _, rule := range repResult.Rule {
		log.Printf("Rule ID: %s", rule.ID)
		if rule.RTC != nil {
			log.Printf("RTC: %s", *rule.RTC)
		}
		if rule.PrefixSet != nil {
			for _, prefix := range rule.PrefixSet.Prefix {
				log.Printf("Prefix: %s", *prefix)
			}
		}
		log.Printf("Action: %s", rule.Action)
		log.Printf("Destination bucket: %s", rule.Destination.Bucket)
		log.Printf("Destination location: %s", rule.Destination.Location)
		log.Printf("Transfer type: %s", rule.Destination.TransferType)
		log.Printf("Status: %s", rule.Status)
		log.Printf("Historical object replication: %s", rule.HistoricalObjectReplication)
		if rule.SyncRole != "" {
			log.Printf("Sync role: %s", rule.SyncRole)
		}
	}
}

Set replication time control (RTC)

Enable or disable RTC on an existing CRR rule. Use the rule ID returned by GetBucketReplication.

package main

import (
	"log"

	"github.com/aliyun/aliyun-oss-go-sdk/oss"
)

func main() {
	provider, err := oss.NewEnvironmentVariableCredentialsProvider()
	if err != nil {
		log.Fatalf("Failed to create credentials provider: %v", err)
	}

	clientOptions := []oss.ClientOption{oss.SetCredentialsProvider(&provider)}
	clientOptions = append(clientOptions, oss.Region("yourRegion"))
	clientOptions = append(clientOptions, oss.AuthVersion(oss.AuthV4))
	client, err := oss.New("yourEndpoint", "", "", clientOptions...)
	if err != nil {
		log.Fatalf("Failed to create OSS client: %v", err)
	}

	bucketName := "srcexamplebucket"

	// Set to "enabled" to turn on RTC, or "disabled" to turn it off.
	enabled := "enabled"

	// Use the rule ID from GetBucketReplication.
	id := "564df6de-7372-46dc-b4eb-10f****"

	rtc := oss.PutBucketRTC{
		RTC: &enabled,
		ID:  id,
	}

	err = client.PutBucketRTC(bucketName, rtc)
	if err != nil {
		log.Fatalf("Failed to put bucket RTC: %v", err)
	}

	log.Println("Put Bucket RTC Success!")
}

Query available destination regions

GetBucketReplicationLocation returns the regions to which data in the source bucket can be replicated, along with the supported transfer types and RTC-eligible regions.

package main

import (
	"encoding/xml"
	"log"

	"github.com/aliyun/aliyun-oss-go-sdk/oss"
)

func main() {
	provider, err := oss.NewEnvironmentVariableCredentialsProvider()
	if err != nil {
		log.Fatalf("Failed to create credentials provider: %v", err)
	}

	clientOptions := []oss.ClientOption{oss.SetCredentialsProvider(&provider)}
	clientOptions = append(clientOptions, oss.Region("yourRegion"))
	clientOptions = append(clientOptions, oss.AuthVersion(oss.AuthV4))
	client, err := oss.New("yourEndpoint", "", "", clientOptions...)
	if err != nil {
		log.Fatalf("Failed to create OSS client: %v", err)
	}

	bucketName := "srcexamplebucket"

	stringData, err := client.GetBucketReplicationLocation(bucketName)
	if err != nil {
		log.Fatalf("Failed to get bucket replication location: %v", err)
	}

	var repLocation oss.GetBucketReplicationLocationResult
	err = xml.Unmarshal([]byte(stringData), &repLocation)
	if err != nil {
		log.Fatalf("Failed to unmarshal XML response: %v", err)
	}

	// Available destination regions.
	for _, location := range repLocation.Location {
		log.Printf("Location: %s", location)
	}

	// Supported transfer types per region.
	for _, transferType := range repLocation.LocationTransferType {
		log.Printf("Location: %s, Transfer types: %s", transferType.Location, transferType.TransferTypes)
	}

	// Regions that support RTC.
	for _, rtcLocation := range repLocation.RTCLocation {
		log.Printf("RTC-eligible location: %s", rtcLocation)
	}

	log.Println("Get Bucket Replication Location Success!")
}

Query replication progress

OSS tracks two types of replication progress:

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

GetBucketReplicationProgress returns both values for a specific rule. Use the rule ID from GetBucketReplication.

package main

import (
	"encoding/xml"
	"log"

	"github.com/aliyun/aliyun-oss-go-sdk/oss"
)

func main() {
	provider, err := oss.NewEnvironmentVariableCredentialsProvider()
	if err != nil {
		log.Fatalf("Failed to create credentials provider: %v", err)
	}

	clientOptions := []oss.ClientOption{oss.SetCredentialsProvider(&provider)}
	clientOptions = append(clientOptions, oss.Region("yourRegion"))
	clientOptions = append(clientOptions, oss.AuthVersion(oss.AuthV4))
	client, err := oss.New("yourEndpoint", "", "", clientOptions...)
	if err != nil {
		log.Fatalf("Failed to create OSS client: %v", err)
	}

	bucketName := "srcexamplebucket"
	ruleId := "564df6de-7372-46dc-b4eb-10f****"

	stringData, err := client.GetBucketReplicationProgress(bucketName, ruleId)
	if err != nil {
		log.Fatalf("Failed to get bucket replication progress: %v", err)
	}

	var repProgress oss.GetBucketReplicationProgressResult
	err = xml.Unmarshal([]byte(stringData), &repProgress)
	if err != nil {
		log.Fatalf("Failed to unmarshal XML response: %v", err)
	}

	for _, rule := range repProgress.Rule {
		log.Printf("Rule ID: %s", rule.ID)
		if rule.PrefixSet != nil {
			for _, prefix := range rule.PrefixSet.Prefix {
				log.Printf("Prefix: %s", *prefix)
			}
		}
		log.Printf("Action: %s", rule.Action)
		log.Printf("Destination bucket: %s", rule.Destination.Bucket)
		log.Printf("Destination location: %s", rule.Destination.Location)
		log.Printf("Transfer type: %s", rule.Destination.TransferType)
		log.Printf("Status: %s", rule.Status)
		log.Printf("Historical object replication: %s", rule.HistoricalObjectReplication)
		if rule.Progress != nil && rule.Progress.HistoricalObject != "" {
			// Percentage of historical objects replicated.
			log.Printf("Historical object replication progress: %s", rule.Progress.HistoricalObject)
		}
		// Timestamp up to which incremental objects have been replicated.
		log.Printf("Incremental replication progress: %s", rule.Progress.NewObject)
	}

	log.Println("Get Bucket Replication Progress Success!")
}

Disable data replication

You can disable the replication relationship between the source and destination buckets by deleting the replication rule of the source bucket. After deletion, OSS stops replicating new operations.

Use the rule ID returned by GetBucketReplication.

package main

import (
	"log"

	"github.com/aliyun/aliyun-oss-go-sdk/oss"
)

func main() {
	provider, err := oss.NewEnvironmentVariableCredentialsProvider()
	if err != nil {
		log.Fatalf("Failed to create credentials provider: %v", err)
	}

	clientOptions := []oss.ClientOption{oss.SetCredentialsProvider(&provider)}
	clientOptions = append(clientOptions, oss.Region("yourRegion"))
	clientOptions = append(clientOptions, oss.AuthVersion(oss.AuthV4))
	client, err := oss.New("yourEndpoint", "", "", clientOptions...)
	if err != nil {
		log.Fatalf("Failed to create OSS client: %v", err)
	}

	srcbucketName := "yourSourceBucket"

	// Use the rule ID returned by GetBucketReplication.
	ruleID := "e047ce28-6806-4131-b1da-30142116****"

	err = client.DeleteBucketReplication(srcbucketName, ruleID)
	if err != nil {
		log.Fatalf("Failed to delete bucket replication: %v", err)
	}

	log.Println("Delete Bucket Replication Success!")
}

Data replication (Go SDK V1)