Elastic Search Source

README.md

@@ -35,6 +35,26 @@ For any other action a warning entry is added to log and Record is skipped.
 | `bulkSize`               | The number of items stored in bulk in the index. The minimum value is `1`, maximum value is `10000`. Note that values greater than `1000` may require additional service configuration.                                                          | `true`                                               | `"1000"` |
 | `retries`                | The maximum number of retries of failed operations. The minimum value is `0` which disabled retry logic. The maximum value is `255`. Note that the higher value, the longer it may take to process retries, as a result, ingest next operations. | `true`                                               | `"1000"` |
 
+
+# Source
+ElasticSearch source connector allows you to move data from multiple Elasticsearch indexes with the specified `host` and `indexes`. It uses elasticsearch search api to pull data from indexes. Upon starting it pulls batches of data from indexes, once all the data is retrieved, it then polls the search api to pull data at regular intervals. 
+
+## Configuration Options
+| name                     | description                                                                                                                                                                                                                                      | required                                             | default  |
+|--------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|------------------------------------------------------|----------|
+| `version`                | The version of the Elasticsearch service. Currently supports version `8`.                                                                                                                                                                           | `true`                                               |          |
+| `host`                   | The Elasticsearch host and port (e.g.: http://127.0.0.1:9200).                                                                                                                                                                                   | `true`                                               |          |
+| `username`               | [v: 5, 6, 7, 8] The username for HTTP Basic Authentication.                                                                                                                                                                                      | `false`                                              |          |
+| `password`               | [v: 5, 6, 7, 8] The password for HTTP Basic Authentication.                                                                                                                                                                                      | `true` when username was provided, `false` otherwise |          |
+| `cloudId`                | [v: 6, 7, 8] Endpoint for the Elastic Service (https://elastic.co/cloud).                                                                                                                                                                        | `false`                                              |          |
+| `apiKey`                 | [v: 6, 7, 8] Base64-encoded token for authorization; if set, overrides username/password and service token.                                                                                                                                      | `false`                                              |          |
+| `serviceToken`           | [v: 7, 8] Service token for authorization; if set, overrides username/password.                                                                                                                                                                  | `false`                                              |          |
+| `certificateFingerprint` | [v: 7, 8] SHA256 hex fingerprint given by Elasticsearch on first launch.                                                                                                                                                                         | `false`                                              |          |
+| `indexes`                  | The name of the indexes to read the data from.                                                                                                                                                                                                      | `true`                                               |          |
+| `batchSize`               | The number of items to fetch from an index. The minimum value is `1`, maximum value is `10000`.                                                          | `false`                                               | `"1000"` |
+| `pollingPeriod`                | The duration for polling the search api for fetching new records. | `false`                                               | `"5s"` |
+
+
 # Testing
 
 Run `make test` to run all the unit and integration tests, which require Docker to be installed and running. The command will handle starting and stopping docker containers for you.
@@ -57,5 +77,6 @@ docker-compose -f test/docker-compose.v8.overrides.yml -f test/docker-compose.v8
 
 - https://github.com/elastic/go-elasticsearch
 - https://www.elastic.co/guide/en/elasticsearch/reference/7.17/docs-bulk.html
+- https://www.elastic.co/guide/en/elasticsearch/reference/current/search-your-data.html
 
 ![scarf pixel](https://static.scarf.sh/a.png?x-pxid=715ebf4a-148c-44ad-8f64-6cc5780d34ae)

connector.go

@@ -16,11 +16,12 @@ package elasticsearch
 
 import (
 	"github.com/conduitio-labs/conduit-connector-elasticsearch/destination"
+	"github.com/conduitio-labs/conduit-connector-elasticsearch/source"
 	sdk "github.com/conduitio/conduit-connector-sdk"
 )
 
 var Connector = sdk.Connector{
 	NewSpecification: Specification,
-	NewSource:        nil,
+	NewSource:        source.NewSource,
 	NewDestination:   destination.NewDestination,
 }

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/elastic/go-elasticsearch/v8 v8.15.0
 	github.com/golangci/golangci-lint v1.61.0
 	github.com/jaswdr/faker v1.19.1
+	github.com/matryer/is v1.4.1
 	github.com/matryer/moq v0.5.0
 	github.com/stretchr/testify v1.9.0
 	go.uber.org/goleak v1.3.0
@@ -128,7 +129,6 @@ require (
 	github.com/maratori/testableexamples v1.0.0 // indirect
 	github.com/maratori/testpackage v1.1.1 // indirect
 	github.com/matoous/godox v0.0.0-20240105082147-c5b5e0e7c0c0 // indirect
-	github.com/matryer/is v1.4.1 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/mattn/go-runewidth v0.0.15 // indirect

internal/elasticsearch/api/search.go

@@ -0,0 +1,36 @@
+// Copyright © 2024 Meroxa, Inc. and Miquido
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package api
+
+// SearchRequest is the request for calling ElasticSearch api.
+type SearchRequest struct {
+	Index       string  `json:"index"`
+	Size        *int    `json:"size:"`
+	SearchAfter []int64 `json:"searchAfter"`
+	SortBy      string  `json:"sortBy"`
+	Order       string  `json:"order"`
+}
+
+// SearchResponse is the JSON response from Elasticsearch search query.
+type SearchResponse struct {
+	Hits struct {
+		Hits []struct {
+			Index  string         `json:"_index"`
+			ID     string         `json:"_id"`
+			Source map[string]any `json:"_source"`
+			Sort   []int64        `json:"sort"` // used for search_after
+		} `json:"hits"`
+	} `json:"hits"`
+}

internal/elasticsearch/client.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"io"
 
+	"github.com/conduitio-labs/conduit-connector-elasticsearch/internal/elasticsearch/api"
 	"github.com/conduitio/conduit-commons/opencdc"
 )
 
@@ -39,4 +40,7 @@ type Client interface {
 
 	// PrepareDeleteOperation prepares delete operation definition for Bulk API query.
 	PrepareDeleteOperation(key string, index string) (metadata interface{}, err error)
+
+	// Search calls the elasticsearch search api and retuns SearchResponse read from an index.
+	Search(ctx context.Context, request *api.SearchRequest) (*api.SearchResponse, error)
 }

internal/elasticsearch/v5/client.go

@@ -21,7 +21,9 @@ import (
 	"fmt"
 	"io"
 
+	"github.com/conduitio-labs/conduit-connector-elasticsearch/internal/elasticsearch/api"
 	"github.com/conduitio/conduit-commons/opencdc"
+
 	"github.com/elastic/go-elasticsearch/v5"
 )
 
@@ -159,3 +161,9 @@ func preparePayload(item *opencdc.Record) (json.RawMessage, error) {
 		return itemPayload.Bytes(), nil
 	}
 }
+
+// Search calls the elasticsearch search api and retuns SearchResponse read from an index.
+func (c *Client) Search(_ context.Context, _ *api.SearchRequest) (*api.SearchResponse, error) {
+	// TODO: implement elasticsearch search api
+	return nil, fmt.Errorf("method not implemented")
+}

internal/elasticsearch/v6/client.go

@@ -21,7 +21,9 @@ import (
 	"fmt"
 	"io"
 
+	"github.com/conduitio-labs/conduit-connector-elasticsearch/internal/elasticsearch/api"
 	"github.com/conduitio/conduit-commons/opencdc"
+
 	"github.com/elastic/go-elasticsearch/v6"
 )
 
@@ -162,3 +164,9 @@ func preparePayload(item *opencdc.Record) (json.RawMessage, error) {
 		return itemPayload.Bytes(), nil
 	}
 }
+
+// Search calls the elasticsearch search api and retuns SearchResponse read from an index.
+func (c *Client) Search(_ context.Context, _ *api.SearchRequest) (*api.SearchResponse, error) {
+	// TODO: implement elasticsearch search api
+	return nil, fmt.Errorf("method not implemented")
+}

internal/elasticsearch/v7/client.go

@@ -21,7 +21,9 @@ import (
 	"fmt"
 	"io"
 
+	"github.com/conduitio-labs/conduit-connector-elasticsearch/internal/elasticsearch/api"
 	"github.com/conduitio/conduit-commons/opencdc"
+
 	"github.com/elastic/go-elasticsearch/v7"
 )
 
@@ -161,3 +163,9 @@ func preparePayload(item *opencdc.Record) (json.RawMessage, error) {
 		return itemPayload.Bytes(), nil
 	}
 }
+
+// Search calls the elasticsearch search api and retuns SearchResponse read from an index.
+func (c *Client) Search(_ context.Context, _ *api.SearchRequest) (*api.SearchResponse, error) {
+	// TODO: implement elasticsearch search api
+	return nil, fmt.Errorf("method not implemented")
+}

internal/elasticsearch/v8/search.go

@@ -0,0 +1,82 @@
+// Copyright © 2022 Meroxa, Inc. and Miquido
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package v8
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"strings"
+	"time"
+
+	"github.com/conduitio-labs/conduit-connector-elasticsearch/internal/elasticsearch/api"
+
+	"github.com/elastic/go-elasticsearch/v8/esapi"
+)
+
+// Search calls the elasticsearch search api and retuns SearchResponse read from an index.
+func (c *Client) Search(ctx context.Context, request *api.SearchRequest) (*api.SearchResponse, error) {
+	// Create the search request
+	req := esapi.SearchRequest{
+		Index: []string{request.Index},
+		Body:  strings.NewReader(createSearchBody(request.SearchAfter)),
+		Size:  request.Size,
+	}
+
+	// Perform the request
+	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	res, err := req.Do(ctx, c.es)
+	if err != nil {
+		return nil, fmt.Errorf("error getting search response: %w", err)
+	}
+	defer res.Body.Close()
+
+	if res.IsError() {
+		return nil, fmt.Errorf("error search response: %s", res.String())
+	}
+
+	var response *api.SearchResponse
+	if err := json.NewDecoder(res.Body).Decode(&response); err != nil {
+		return nil, fmt.Errorf("error parsing the search response body: %w", err)
+	}
+
+	return response, nil
+}
+
+func createSearchBody(searchAfter []int64) string {
+	body := map[string]interface{}{
+		"query": map[string]interface{}{
+			"match_all": struct{}{},
+		},
+		"sort": []map[string]interface{}{
+			{"_seq_no": map[string]string{
+				"order": "asc",
+			}},
+		},
+	}
+
+	if len(searchAfter) == 1 {
+		body["search_after"] = searchAfter
+	}
+
+	jsonBody, err := json.Marshal(body)
+	if err != nil {
+		log.Printf("error marshaling the search request body: %s", err)
+	}
+
+	return string(jsonBody)
+}