【原创】从s3桶将对象导入ES建立索引，以便快速查找文件
总体功能：
这段程序的作用是：
从指定的S3桶中读取所有对象的元数据（文件名、大小、最后修改时间、存储类型、ETag等），并把这些信息写入到Elasticsearch（ES）中，建立索引，以便后续可以快速搜索和检索S3中的文件信息。
各模块详细解析：
1. readConfig
-
从本地的
config.ini配置文件中读取 S3 和 Elasticsearch 的连接信息，比如 S3的桶名、访问秘钥，ES的地址、账号密码、索引名等。
2. getS3ETag
-
给定一个文件名（Key），向S3发送
HeadObject请求，单独获取该文件的ETag（一般用于文件完整性校验或者去重标识）。 -
特别处理了返回的ETag，把多余的引号去掉。
3. fetchS3Files
-
核心逻辑
-
通过 S3 的
ListObjectsV2Paginator，分页地遍历桶中的所有对象。 -
对每一个对象：
-
读取文件名（
Key）、大小、最后修改时间、存储类型。 -
调用
getS3ETag补充获取 ETag。 -
把这些元数据组织成一个
JSON格式的文档。 -
通过
esapi.IndexRequest把文档写入到Elasticsearch的指定索引中。
-
4. main
-
先读取配置文件。
-
初始化 S3 客户端（带自定义 Endpoint，例如MinIO、私有S3等），并提供静态访问密钥。
-
初始化 Elasticsearch 客户端（支持跳过TLS证书验证，适合开发环境）。
-
调用
fetchS3Files，正式开始批量导入S3文件元数据到ES。
特别注意：
-
InsecureSkipVerify: true表示 跳过SSL证书校验，这个设置在生产环境是不安全的，一般仅用于开发测试。 -
每导入一个S3对象，都会立即
Refresh索引，确保数据立刻可搜索，但也会增加ES压力，批量模式可以更高效。 -
如果S3里文件特别多，可以考虑加并发处理或限制速率，否则会慢。
总结一句话：
这段程序实现了 自动同步S3文件列表到Elasticsearch索引，方便对存储桶中的文件进行快速搜索和查询
package mainimport ("bytes""context""encoding/json""fmt""github.com/aws/aws-sdk-go-v2/aws""github.com/aws/aws-sdk-go-v2/config""github.com/aws/aws-sdk-go-v2/credentials""github.com/aws/aws-sdk-go-v2/service/s3""github.com/elastic/go-elasticsearch/v8""github.com/elastic/go-elasticsearch/v8/esapi""gopkg.in/ini.v1""log""crypto/tls""net/http""time" )type S3Config struct {BucketName stringAccessKey stringSecretKey stringEndpointURL string }type ESConfig struct {Host stringUser stringPass stringIndexName stringSearchType string }func readConfig() (S3Config, ESConfig) {cfg, err := ini.Load("config.ini")if err != nil {log.Fatalf("无法读取配置文件: %v", err)}s3Cfg := S3Config{BucketName: cfg.Section("s3").Key("bucket_name").String(),AccessKey: cfg.Section("s3").Key("access_key").String(),SecretKey: cfg.Section("s3").Key("secret_key").String(),EndpointURL: cfg.Section("s3").Key("endpoint_url").String(),}esCfg := ESConfig{Host: cfg.Section("elasticsearch").Key("host").String(),User: cfg.Section("elasticsearch").Key("user").String(),Pass: cfg.Section("elasticsearch").Key("password").String(),IndexName: cfg.Section("elasticsearch").Key("index_name").String(),SearchType: cfg.Section("elasticsearch").Key("search_type").String(),}return s3Cfg, esCfg }func getS3ETag(s3Client *s3.Client, bucketName, fileKey string) string {resp, err := s3Client.HeadObject(context.TODO(), &s3.HeadObjectInput{Bucket: aws.String(bucketName),Key: aws.String(fileKey),})if err != nil {log.Printf("获取 %s 的ETag失败: %v", fileKey, err)return ""}etag := aws.ToString(resp.ETag)if len(etag) > 0 && etag[0] == '"' && etag[len(etag)-1] == '"' {etag = etag[1 : len(etag)-1]}return etag }func fetchS3Files(s3Client *s3.Client, esClient *elasticsearch.Client, bucketName, indexName string) {paginator := s3.NewListObjectsV2Paginator(s3Client, &s3.ListObjectsV2Input{Bucket: aws.String(bucketName),})for paginator.HasMorePages() {page, err := paginator.NextPage(context.TODO())if err != nil {log.Printf("获取S3文件列表页失败: %v", err)continue}for _, obj := range page.Contents {fileKey := aws.ToString(obj.Key)log.Printf("导入索引：",fileKey)fileSize := aws.ToInt64(obj.Size)lastModified := obj.LastModifiedstorageClass := string(obj.StorageClass) // 修复点etag := getS3ETag(s3Client, bucketName, fileKey)fileData := map[string]interface{}{"file_key": fileKey,"file_size": fileSize,"last_modified": lastModified,"storage_class": storageClass,"etag": etag,}fileDataJSON, err := json.Marshal(fileData)if err != nil {log.Printf("将文件数据转换为JSON失败: %v", err)continue}req := esapi.IndexRequest{Index: indexName,Body: bytes.NewReader(fileDataJSON), // 修复点Refresh: "true",}resp, err := req.Do(context.TODO(), esClient)if err != nil {log.Printf("将文件数据索引到Elasticsearch失败: %v", err)continue}defer resp.Body.Close()}}fmt.Println("S3 文件索引完成") }func main() {s3Cfg, esCfg := readConfig()customResolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {return aws.Endpoint{URL: s3Cfg.EndpointURL,SigningRegion: "us-east-1", // 替换为你的实际regionHostnameImmutable: true,}, nil})awsCfg, err := config.LoadDefaultConfig(context.TODO(),config.WithRegion("us-east-1"), // 替换为你的实际regionconfig.WithEndpointResolverWithOptions(customResolver),config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(s3Cfg.AccessKey,s3Cfg.SecretKey,"",)),)if err != nil {log.Fatalf("无法加载S3配置: %v", err)}s3Client := s3.NewFromConfig(awsCfg)esCfgOptions := elasticsearch.Config{Addresses: []string{esCfg.Host},Username: esCfg.User,Password: esCfg.Pass,Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true, // ⚠️ 跳过证书校验（不安全，仅限开发）},ResponseHeaderTimeout: 10 * time.Second,},}esClient, err := elasticsearch.NewClient(esCfgOptions)if err != nil {log.Fatalf("无法创建Elasticsearch客户端: %v", err)}fetchS3Files(s3Client, esClient, s3Cfg.BucketName, esCfg.IndexName) }
配置文件config.ini
[elasticsearch]
host = https://localhost:9200
user = elastic
password = U********uq
index_name = jyzx_s3_files
search_type = wildcard[s3]
bucket_name = fzsjyzx
access_key = V4*****6DB
secret_key = lHdmyi5*********UjlS
endpoint_url = http://172.20.1.18:7480/