关于前端:ES日志存储以及备份压缩到COS

导语

为了满足用户日益增长的日志存储大小，不影响用户的写入和查问性能。满足不同用户写入流量。同时用户日志长期保留，日志存储比拟占用空间和老本。ES集群规格配置高，耗费资源和老本。咱们基于Go语言设计了一个多用户多ES集群，日志备份到cos节省成本的计划。本篇实际基于Go语言编程。

索引设计

为了避免单个索引一直减少。影响ES集群查问写入性能，ES 集群的索引设计次要采取如下形式：

1. 租户拆散：将索引依照租户进行拆散，防止不同租户之间的数据混同，进步 ES 集群的数据安全性和隔离性。
2. 按月宰割：将索引依照每个月进行宰割，防止单个索引过大，进步 ES 集群的查问性能。
3. 按大小固定rollover：将索引依照固定大小进行rollover，主动分索引，防止单个索引过大，进行索引治理，晋升集群性能，便于集群运维。

ES集群设计

为了防止 ES 集群呈现单点问题，以及配置一直减少带来的运维危险， ES 集群设计时次要通过将不同规格的用户日志写入不同配置的ES集群，并且应用主动扩容技术来实现集群的可扩展性。
具体来说，能够将用户日志依照不同的规格（例如不同的数据量、拜访频率等）进行分类，而后将不同规格的用户日志写入不同配置的ES集群中。例如，能够应用较小的ES集群来解决低频拜访的用户日志，而应用较大的ES集群来解决高频拜访的用户日志。

ES数据备份到COS

创立备份流程
在 ES 备份流程中，为了保障备份的正确性和完整性，能够依照如下步骤执行：
创立一个仓库来存储备份数据。
创立一个快照工作，将ES中的数据备份到指定的仓库中。在创立快照工作时，须要指定备份的索引、仓库、快照名称等参数，并设置备份的间隔时间和保留工夫等策略。
查问快照工作是否实现。ES提供了API接口来查问快照的状态和进度，能够依据查问后果来判断备份是否曾经实现。
调用 COS 提供的压缩函数对备份数据进行压缩，以减小备份数据所占用的存储空间。

通过备份和压缩到 COS 能够将 ES 存储的数据放大到 1/4 的大小，并且通过 COS 的海量分布式存储服务，能够提供低成本、弹性的云存储服务。

各项步骤对应的代码逻辑如下：
1.创立仓库

func CreateRepository(Repository string, cosappid uint64, appId uint64, accessKeyId string, AccessKeySecret string, buckets string, region string) error {  service := Client.SnapshotCreateRepository(Repository)  service = service.Type("cos").Settings(map[string]interface{}{    "app_id":            strconv.FormatUint(cosappid, 10),    "access_key_id":     accessKeyId,    "access_key_secret": AccessKeySecret,    "bucket":            buckets,    "region":            region,    "compress":          true,    "chunk_size":        "500mb",    "base_path":         "backup/" + strconv.FormatUint(appId, 10) + "/" + Repository,  })  if err := service.Validate(); err != nil {    logger.Errorf("SnapshotCreateRepository failed,app_id=%v,Repository=%s ", appId, Repository)    return err  }  _, err := service.Do(context.Background())  if err != nil {    logger.Errorf("SnapshotGetRepository do failed,app_id=%v,Repository=%s ", appId, Repository)  }  return nil}

2.创立快照

func SnapshotCreate(Repository string, Snapshot string, appId uint64, indexname string) error {  type IndicesSetting struct {    Indices string `json:"indices"`  }  e1 := IndicesSetting{indexname}  _, err := Client.SnapshotCreate(Repository, Snapshot).WaitForCompletion(false).BodyJson(e1).Pretty(true).Do(context.Background())  if err != nil {    logger.Errorf("SnapshotCreate failed,app_id=%v,Repository=%s Snapshot=%s ", appId, Repository, Snapshot)    return err  }  return nil}

3.查问快照状态

func SnapshotGet(Repository string, Snapshot string, appId uint64) (*elastic.SnapshotGetResponse, error) {  service, err := Client.SnapshotGet(Repository).    Snapshot(Snapshot).    IgnoreUnavailable(true).    Verbose(true).Pretty(true).Do(context.Background())  if err != nil {    logger.Errorf("SnapshotGet failed,app_id=%v,Repository=%s Snapshot=%s ", appId, Repository, Snapshot)    return nil, err  } /*    判断工作是否实现    if service.Snapshots[0].State != "true" {    }*/  return service, nil}

4.创立压缩工作

压缩工作是腾讯云对象存储 COS 提供的压缩API，须要先创立好压缩函数：
创立压缩函数参考如下：
https://cloud.tencent.com/document/product/436/58578

首先获取压缩文件，而后调用压缩函数进行压缩。获取压缩文件代码逻辑如下：

var coslist []string  coslist, err = ext.GetCosDirCosKeyList(backupcosfile, cosClient)  if err != nil {    logger.Errorx("GetCosDirCosKeyList error", err)    return err  }  scfClient, err := ext.GetScfClient(config.One.Backup.Region, true)  if err != nil {    logger.Errorx("GetScfClient error", err)    return err  }  var invokeRequestId string  invokeRequestId, err = ext.ScfCompression(scfClient, coslist, backupcosfilezip, config.One.Backup.ZipFunctionName)  if err != nil {    logger.Errorx("GetScfClient error", err)    return err  }

压缩函数模版如下：

func ScfCompression(scfCleint *scf.Client, cosKeyList []string, compressKey string, functionname string) (functionRequestId string, err error) {  request := scf.NewInvokeRequest()  request.FunctionName = common.StringPtr(functionname)  scfSourceList := make([]ScfCompressionSource, 0, 1)  for _, cosKey := range cosKeyList {    scfSource := ScfCompressionSource{      Url: fmt.Sprintf("https://%s.cos-internal.%s.tencentcos.cn/%s", config.One.Cos.GuangzhouCosBucket, config.One.Cos.CosRegion, cosKey),    }    scfSourceList = append(scfSourceList, scfSource)  }  scfClientContext := ScfCompressionClientContext{    Bucket:     config.One.Cos.GuangzhouCosBucket,    Region:     config.One.Cos.CosRegion,    Key:        compressKey,    Flatten:    false,    SourceList: scfSourceList,  }  clientContext, err := json.Marshal(scfClientContext)  if err != nil {    logger.Errorx("ScfCompression Marshal", err)    return "", err  }  request.ClientContext = common.StringPtr(string(clientContext))  //fmt.Println(string(clientContext))  response, err := scfCleint.Invoke(request)  if _, ok := err.(*tencentError.TencentCloudSDKError); ok {    logger.Errorx("An API error has returned:", err)    return "", err  }  if err != nil {    logger.Errorx("scfCleint.Invoke", err)    return "", err  }  logger.Debugf("%s", response.ToJsonString())  return *response.Response.Result.FunctionRequestId, nil}

COS数据恢复到ES

创立复原流程
复原流程是备份流程的逆向流程，次要包含如下步骤：
创立COS解压缩函数
创立复原工作
查问复原工作是否实现
实现复原

1. COS 解压缩工作
   解压缩工作是腾讯云对象存储 COS 提供的解压缩函数模板，须要先创立好解压缩函数：

创立解压函数参考如下：
https://cloud.tencent.com/document/product/436/67101

解压缩函数模板如下：

func ScfDecompression(scfCleint *scf.Client, compressKey string, cosTargetPrefix string, functionname string) (functionRequestId string, err error) {  logger.Debug("ScfDecompression" + compressKey)  request := scf.NewInvokeRequest()  request.FunctionName = common.StringPtr(functionname)  //key := fmt.Sprintf("https://%s.cos-internal.ap-guangzhou.myqcloud.com/%s", config.One.Cos.GuangzhouCosBucket, compressKey)  scfClientContext := ScfDecompressionClientContext{    Bucket:       config.One.Cos.GuangzhouCosBucket,    Region:       config.One.Cos.CosRegion,    Key:          compressKey,    TargetBucket: config.One.Cos.GuangzhouCosBucket,    TargetRegion: config.One.Cos.CosRegion,    TargetPrefix: cosTargetPrefix,  }  clientContext, err := json.Marshal(scfClientContext)  if err != nil {    logger.Errorx("ScfCompression Marshal", err)    return "", err  }  request.ClientContext = common.StringPtr(string(clientContext))  //fmt.Println(string(clientContext))  response, err := scfCleint.Invoke(request)  if _, ok := err.(*tencentError.TencentCloudSDKError); ok {    logger.Errorx("An API error has returned:", err)    return "", err  }  if err != nil {    logger.Errorx("scfCleint.Invoke", err)    return "", err  }  logger.Debugf("%s", response.ToJsonString())  return *response.Response.Result.FunctionRequestId, nil}

2. 实现解压缩后，创立复原工作

func SnapshotRestore(Repository string, Snapshot string, appId uint64, oldIndexName string, newIndexName string) error {  _, err := Client.SnapshotRestore(Repository, Snapshot).RenamePattern(oldIndexName).    RenameReplacement(newIndexName).Indices(oldIndexName).WaitForCompletion(false).Pretty(true).Do(context.Background())  if err != nil {    logger.Errorf("SnapshotRestore failed,app_id=%v,Repository=%s Snapshot=%s ", appId, Repository, Snapshot)    return err  }  return nil}