Merge pull request #5 from trifork/feature/opensearch-cache

Feature/opensearch cache

internal/impl/opensearch/cache.go

@@ -73,13 +73,18 @@ type opensearchCache struct {
 	valueField string
 }
 
-func newOpensearchCache(indexName, keyField, valueField string, clientOpts opensearchapi.Config) (*opensearchCache, error) {
+func newOpensearchCache(indexName string, keyField string, valueField string, clientOpts opensearchapi.Config) (*opensearchCache, error) {
 
 	client, err := opensearchapi.NewClient(clientOpts)
 	if err != nil {
 		return nil, err
 	}
 
+	_, err = client.Ping(context.Background(), &opensearchapi.PingReq{})
+	if err != nil {
+		return nil, fmt.Errorf("error pinging opensearch: %v", err)
+	}
+
 	return &opensearchCache{
 		client:     client,
 		indexName:  indexName,
@@ -126,7 +131,6 @@ func (m *opensearchCache) Get(ctx context.Context, key string) ([]byte, error) {
 		}
 
 		searchResponse, err := m.client.Search(ctx, search)
-
 		if err != nil {
 			return nil, fmt.Errorf("error searching for key %s: %v", key, err)
 		}

resources/docker/opensearch/cache.yaml

@@ -1,29 +1,85 @@
+logger:
+  level: VERBOSE
+
 pipeline:
   processors:
+    - label: doclevel_cache
+      cache:
+        resource: opensearch_entiredoc
+        operator: get
+        key: "1"
     - label: fieldvalue_cache
       cache:
-        resource: opensearch
+        resource: opensearch_fieldvalue
         operator: get
         key: "1"
+    - label: leveled_cache
+      processors:
+        - cache:
+            resource: opensearch_entiredoc
+            operator: get
+            key: "1"
+        - cache:
+            resource: hot
+            operator: get
+            key: "1"
+        - cached:
+            key: "1"
+            cache: hot
+            processors:
+              - mapping: throw("hot cache not found")
 
 cache_resources:
-  - label: opensearch
-    opensearch: 
+  - label: opensearch_entiredoc
+    opensearch:
       urls: ["https://osproxy-saptest.cheetah.trifork.dev"]
       index: sap-salesorders_s4hanacloud
       key_field: SalesOrder
       value_field: "" # CreatedByUser
       basic_auth:
         enabled: true
-        username: "apikey"
-        password: "${OS_API_KEY}"
+        username: ${USERNAME}
+        password: ${PASSWORD}
+  - label: opensearch_fieldvalue
+    opensearch:
+      urls: ["https://osproxy-saptest.cheetah.trifork.dev"]
+      index: sap-salesorders_s4hanacloud
+      key_field: SalesOrder
+      value_field: "CreatedByUser"
+      basic_auth:
+        enabled: true
+        username: ${USERNAME}
+        password: ${PASSWORD}
+  - label: leveled
+    multilevel: [hot, opensearch_entiredoc]
+  - label: hot
+    memory:
+      default_ttl: 60s
 
 tests:
+  - name: Example test case doclevel_cache
+    environment: {}
+    target_processors: doclevel_cache #/pipeline/processors
+    input_batch:
+      - content: "ignored value"
+    output_batches:
+      - - json_contains:
+            CreatedByUser: "CB9980000027"
   - name: Example test case fieldvalue_cache
     environment: {}
     target_processors: fieldvalue_cache #/pipeline/processors
     input_batch:
-      - content: 'ignored value'
+      - content: "ignored value"
+    output_batches:
+      - - content_equals: "CB9980000027"
+  - name: Example test case leveled_cache
+    environment: {}
+    target_processors: leveled_cache #/pipeline/processors
+    input_batch:
+      - content: "ignored value"
+      - content: "ignored value"
     output_batches:
-      - - json_contains: 
-            CreatedByUser: 'CB9980000027'
+      - - json_contains:
+            CreatedByUser: "CB9980000027"
+        - json_contains:
+            CreatedByUser: "CB9980000027"

resources/docker/opensearch/docker-compose.yaml

@@ -18,3 +18,6 @@ services:
     volumes:
       - ./cache.yaml:/benthos.yaml
     network_mode: host
+    environment:
+      - USERNAME=apikey
+      - PASSWORD=${OS_API_KEY}