123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259 |
- package minio
- import (
- "bytes"
- "context"
- "encoding/json"
- "encoding/xml"
- "io/ioutil"
- "net/http"
- "net/url"
- "time"
- "github.com/google/uuid"
- "github.com/minio/minio-go/v7/pkg/replication"
- "github.com/minio/minio-go/v7/pkg/s3utils"
- )
- func (c Client) RemoveBucketReplication(ctx context.Context, bucketName string) error {
- return c.removeBucketReplication(ctx, bucketName)
- }
- func (c Client) SetBucketReplication(ctx context.Context, bucketName string, cfg replication.Config) error {
-
- if err := s3utils.CheckValidBucketName(bucketName); err != nil {
- return err
- }
-
- if cfg.Empty() {
- return c.removeBucketReplication(ctx, bucketName)
- }
-
- return c.putBucketReplication(ctx, bucketName, cfg)
- }
- func (c Client) putBucketReplication(ctx context.Context, bucketName string, cfg replication.Config) error {
-
-
- urlValues := make(url.Values)
- urlValues.Set("replication", "")
- replication, err := xml.Marshal(cfg)
- if err != nil {
- return err
- }
- reqMetadata := requestMetadata{
- bucketName: bucketName,
- queryValues: urlValues,
- contentBody: bytes.NewReader(replication),
- contentLength: int64(len(replication)),
- contentMD5Base64: sumMD5Base64(replication),
- }
-
- resp, err := c.executeMethod(ctx, http.MethodPut, reqMetadata)
- defer closeResponse(resp)
- if err != nil {
- return err
- }
- if resp.StatusCode != http.StatusOK {
- return httpRespToErrorResponse(resp, bucketName, "")
- }
- return nil
- }
- func (c Client) removeBucketReplication(ctx context.Context, bucketName string) error {
-
-
- urlValues := make(url.Values)
- urlValues.Set("replication", "")
-
- resp, err := c.executeMethod(ctx, http.MethodDelete, requestMetadata{
- bucketName: bucketName,
- queryValues: urlValues,
- contentSHA256Hex: emptySHA256Hex,
- })
- defer closeResponse(resp)
- if err != nil {
- return err
- }
- return nil
- }
- func (c Client) GetBucketReplication(ctx context.Context, bucketName string) (cfg replication.Config, err error) {
-
- if err := s3utils.CheckValidBucketName(bucketName); err != nil {
- return cfg, err
- }
- bucketReplicationCfg, err := c.getBucketReplication(ctx, bucketName)
- if err != nil {
- errResponse := ToErrorResponse(err)
- if errResponse.Code == "ReplicationConfigurationNotFoundError" {
- return cfg, nil
- }
- return cfg, err
- }
- return bucketReplicationCfg, nil
- }
- func (c Client) getBucketReplication(ctx context.Context, bucketName string) (cfg replication.Config, err error) {
-
-
- urlValues := make(url.Values)
- urlValues.Set("replication", "")
-
- resp, err := c.executeMethod(ctx, http.MethodGet, requestMetadata{
- bucketName: bucketName,
- queryValues: urlValues,
- })
- defer closeResponse(resp)
- if err != nil {
- return cfg, err
- }
- if resp.StatusCode != http.StatusOK {
- return cfg, httpRespToErrorResponse(resp, bucketName, "")
- }
- if err = xmlDecoder(resp.Body, &cfg); err != nil {
- return cfg, err
- }
- return cfg, nil
- }
- func (c Client) GetBucketReplicationMetrics(ctx context.Context, bucketName string) (s replication.Metrics, err error) {
-
- if err := s3utils.CheckValidBucketName(bucketName); err != nil {
- return s, err
- }
-
-
- urlValues := make(url.Values)
- urlValues.Set("replication-metrics", "")
-
- resp, err := c.executeMethod(ctx, http.MethodGet, requestMetadata{
- bucketName: bucketName,
- queryValues: urlValues,
- })
- defer closeResponse(resp)
- if err != nil {
- return s, err
- }
- if resp.StatusCode != http.StatusOK {
- return s, httpRespToErrorResponse(resp, bucketName, "")
- }
- respBytes, err := ioutil.ReadAll(resp.Body)
- if err != nil {
- return s, err
- }
- if err := json.Unmarshal(respBytes, &s); err != nil {
- return s, err
- }
- return s, nil
- }
- func mustGetUUID() string {
- u, err := uuid.NewRandom()
- if err != nil {
- return ""
- }
- return u.String()
- }
- func (c Client) ResetBucketReplication(ctx context.Context, bucketName string, olderThan time.Duration) (rID string, err error) {
- rID = mustGetUUID()
- _, err = c.resetBucketReplicationOnTarget(ctx, bucketName, olderThan, "", rID)
- if err != nil {
- return rID, err
- }
- return rID, nil
- }
- func (c Client) ResetBucketReplicationOnTarget(ctx context.Context, bucketName string, olderThan time.Duration, tgtArn string) (rinfo replication.ResyncTargetsInfo, err error) {
- rID := mustGetUUID()
- return c.resetBucketReplicationOnTarget(ctx, bucketName, olderThan, tgtArn, rID)
- }
- func (c Client) resetBucketReplicationOnTarget(ctx context.Context, bucketName string, olderThan time.Duration, tgtArn string, resetID string) (rinfo replication.ResyncTargetsInfo, err error) {
-
- if err = s3utils.CheckValidBucketName(bucketName); err != nil {
- return
- }
-
-
- urlValues := make(url.Values)
- urlValues.Set("replication-reset", "")
- if olderThan > 0 {
- urlValues.Set("older-than", olderThan.String())
- }
- if tgtArn != "" {
- urlValues.Set("arn", tgtArn)
- }
- urlValues.Set("reset-id", resetID)
-
- resp, err := c.executeMethod(ctx, http.MethodPut, requestMetadata{
- bucketName: bucketName,
- queryValues: urlValues,
- })
- defer closeResponse(resp)
- if err != nil {
- return rinfo, err
- }
- if resp.StatusCode != http.StatusOK {
- return rinfo, httpRespToErrorResponse(resp, bucketName, "")
- }
- respBytes, err := ioutil.ReadAll(resp.Body)
- if err != nil {
- return rinfo, err
- }
- if err := json.Unmarshal(respBytes, &rinfo); err != nil {
- return rinfo, err
- }
- return rinfo, nil
- }
|