openSearchClient

package
v1.3.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Mar 21, 2024 License: GPL-3.0 Imports: 24 Imported by: 0

README

openSearchClient

import "github.com/greenbone/opensight-golang-libraries/pkg/openSearch/openSearchClient"

Package openSearchClient provides a client for OpenSearch designed to allow easy mocking in tests.

Example Usage:

clientConfig, err := config.ReadOpensearchClientConfig()
if err != nil {
	return err
}

opensearchProjectClient, err := NewOpenSearchProjectClient(context.Background(), clientConfig)
if err != nil {
	return err
}

client := NewClient(opensearchProjectClient, 10, 1)

query := `{"query":{"bool":{"filter":[{"term":{"oid":{"value":"1.3.6.1.4.1.25623.1.0.117842"}}}]}}}`
responseBody, err := client.Search(indexName, []byte(query))
if err != nil {
	return err
}

searchResponse, err := UnmarshalSearchResponse[*Vulnerability](responseBody)
if err != nil {
	return err
}

For further usage examples see ./client_test.go.

Index

func GetResponseError

func GetResponseError(statusCode int, responseString []byte, indexName string) error

GetResponseError checks if a response from OpenSearch indicated success and returns an error if not.

func InitializeJson

func InitializeJson(timeFormats []string)

func NewOpenSearchErrorWithStack

func NewOpenSearchErrorWithStack(message string) error

func NewOpenSearchProjectClient

func NewOpenSearchProjectClient(ctx context.Context, config config.OpensearchClientConfig) (*opensearch.Client, error)

NewOpenSearchProjectClient creates a new official OpenSearch client (package github.com/opensearch-project/opensearch-go) for usage NewClient. It returns an error if the client couldn't be created or the connection couldn't be established.

ctx is the context to use for the connection. config is the configuration for the client.

func NewOpenSearchResourceAlreadyExistsWithStack

func NewOpenSearchResourceAlreadyExistsWithStack(message string) error

func NewOpenSearchResourceNotFoundWithStack

func NewOpenSearchResourceNotFoundWithStack(message string) error

func SerializeDocumentsForBulkUpdate

func SerializeDocumentsForBulkUpdate[T any](indexName string, documents []T) ([]byte, error)

SerializeDocumentsForBulkUpdate serializes documents for bulk update. Can be used in conjunction with BulkUpdate. It returns the serialized documents or an error in case something went wrong.

indexName is the name of the index to update. documents are the documents to update.

func StartOpensearchTestContainer

func StartOpensearchTestContainer(ctx context.Context) (testcontainers.Container, config.OpensearchClientConfig, error)

StartOpensearchTestContainer starts a test container with opensearch and returns the container and the config for the opensearch client. It returns an error if the container couldn't be created or started.

ctx is the context to use for the container.

func Unmarshal

func Unmarshal(data []byte, v any) error

Unmarshal unmarshalls data into v. It returns an error if the data is invalid.

func UnmarshalWithoutValidation

func UnmarshalWithoutValidation(data []byte, v any) error

UnmarshalWithoutValidation unmarshalls data into v. It returns an error if the data can not be parsed.

type Bucket

type Bucket struct {
    Key         any    `json:"key"`
    KeyAsString string `json:"key_as_string"`
    DocCount    uint   `json:"doc_count"`
    Aggs        map[string]DynamicAggregation
}

func (*Bucket) UnmarshalJSON
func (bucket *Bucket) UnmarshalJSON(bytes []byte) error

type BulkResponse

BulkResponse bulk response

type BulkResponse struct {
    Took     uint         `json:"took"`
    HasError bool         `json:"errors"`
    Errors   []IndexError `json:"items"`
}

type Client

Client is a client for OpenSearch designed to allow easy mocking in tests. It is a wrapper around the official OpenSearch client github.com/opensearch-project/opensearch-go .

type Client struct {
    // contains filtered or unexported fields
}

func NewClient
func NewClient(openSearchProjectClient *opensearch.Client, updateMaxRetries int, updateRetryDelay time.Duration) *Client

NewClient creates a new OpenSearch client.

openSearchProjectClient is the official OpenSearch client to wrap. Use NewOpenSearchProjectClient to create it. updateMaxRetries is the number of retries for update requests. updateRetryDelay is the delay between retries.

func (*Client) AsyncDeleteByQuery
func (c *Client) AsyncDeleteByQuery(indexName string, requestBody []byte) error

AsyncDeleteByQuery updates documents in the given index asynchronously. It does not wait for the update to finish before returning. It returns an error in case something went wrong.

indexName is the name of the index to delete from. requestBody is the request body to send to OpenSearch to identify the documents to be deleted.

func (*Client) BulkUpdate
func (c *Client) BulkUpdate(indexName string, requestBody []byte) error

BulkUpdate performs a bulk update in the given index. It returns an error in case something went wrong.

indexName is the name of the index to update. requestBody is the request body to send to OpenSearch specifying the bulk update.

func (*Client) Close
func (c *Client) Close()

Close stops the underlying UpdateQueue allowing a graceful shutdown.

func (*Client) DeleteByQuery
func (c *Client) DeleteByQuery(indexName string, requestBody []byte) error

DeleteByQuery updates documents in the given index. It waits for the update to finish before returning. It returns an error in case something went wrong.

indexName is the name of the index to delete from. requestBody is the request body to send to OpenSearch to identify the documents to be deleted.

func (c *Client) Search(indexName string, requestBody []byte) (responseBody []byte, err error)

Search searches for documents in the given index.

indexName is the name of the index to search in. requestBody is the request body to send to OpenSearch. It returns the response body as or an error in case something went wrong.

func (*Client) Update
func (c *Client) Update(indexName string, requestBody []byte) (responseBody []byte, err error)

Update updates documents in the given index using UpdateQueue (which is also part of this package). It does not wait for the update to finish before returning. It returns the response body as or an error in case something went wrong.

indexName is the name of the index to update. requestBody is the request body to send to OpenSearch specifying the update.

type CreatedResponse

type CreatedResponse struct {
    Id     string `json:"_id"`
    Result string `json:"result"`
}

type DocumentError

type DocumentError struct {
    IndexName  string            `json:"_index"`
    IndexType  string            `json:"_type"`
    DocumentId string            `json:"_id"`
    StatusCode uint              `json:"status"`
    Error      DocumentErrorType `json:"error"`
}

type DocumentErrorType

type DocumentErrorType struct {
    Type   string `json:"type"`
    Reason string `json:"reason"`
}

type DynamicAggregation

type DynamicAggregation struct {
    DocCountErrorUpperBound int                    `json:"doc_count_error_upper_bound"`
    SumOtherDocCount        uint                   `json:"sum_other_doc_count"`
    Buckets                 []Bucket               `json:"buckets"`
    Value                   any                    `json:"value"`
    ValueAsString           any                    `json:"value_as_string"`
    Hits                    DynamicAggregationHits `json:"hits"`
}

type DynamicAggregationHits

type DynamicAggregationHits struct {
    Total      SearchResponseHitsTotal `json:"total"`
    SearchHits KeepJsonAsString        `json:"hits"`
}

type IndexError

type IndexError struct {
    Index DocumentError `json:"index"`
}

type KeepJsonAsString

type KeepJsonAsString []byte

func (*KeepJsonAsString) UnmarshalJSON
func (k *KeepJsonAsString) UnmarshalJSON(data []byte) error

type OpenSearchError

OpenSearchError openSearch error

type OpenSearchError struct {
    Message string
}

func NewOpenSearchError
func NewOpenSearchError(message string) *OpenSearchError

func (*OpenSearchError) Error
func (o *OpenSearchError) Error() string

type OpenSearchErrorResponse

type OpenSearchErrorResponse struct {
    Error  OpenSearchErrors
    Status int
}

type OpenSearchErrors

type OpenSearchErrors struct {
    Reasons []OpenSearchRootCause
    Type    string
    Reason  string
}

type OpenSearchResourceAlreadyExists

OpenSearchResourceAlreadyExists openSearch resource already exists

type OpenSearchResourceAlreadyExists struct {
    Message string
}

func NewOpenSearchResourceAlreadyExists
func NewOpenSearchResourceAlreadyExists(message string) *OpenSearchResourceAlreadyExists

func (*OpenSearchResourceAlreadyExists) Error
func (o *OpenSearchResourceAlreadyExists) Error() string

type OpenSearchResourceNotFound

OpenSearchResourceNotFound openSearch resource already exists

type OpenSearchResourceNotFound struct {
    Message string
}

func NewOpenSearchResourceNotFound
func NewOpenSearchResourceNotFound(message string) *OpenSearchResourceNotFound

func (*OpenSearchResourceNotFound) Error
func (o *OpenSearchResourceNotFound) Error() string

type OpenSearchRootCause

type OpenSearchRootCause struct {
    Type   string
    Reason string
}

type OpensearchTestContainer

OpensearchTestContainer represents the opensearch container

type OpensearchTestContainer struct {
    testcontainers.Container
}

type Request

type Request struct {
    IndexName   string
    RequestBody []byte
    Response    chan Response // Use the new Response type
}

type Response

type Response struct {
    Body []byte
    Err  error
}

type SearchResponse

type SearchResponse[T any] struct {
    Took         uint                       `json:"took"`
    TimedOut     bool                       `json:"timed_out"`
    Hits         SearchResponseHits[T]      `json:"hits"`
    Aggregations SearchResponseAggregations `json:"aggregations"`
}

func UnmarshalSearchResponse
func UnmarshalSearchResponse[T any](data []byte) (*SearchResponse[T], error)

func (SearchResponse[T]) GetResults
func (s SearchResponse[T]) GetResults() []T

GetResults returns list of documents

func (SearchResponse[T]) GetSearchHits
func (s SearchResponse[T]) GetSearchHits() []SearchResponseHit[T]

type SearchResponseAggregation

type SearchResponseAggregation struct {
    DocCountErrorUpperBound int      `json:"doc_count_error_upper_bound"`
    SumOtherDocCount        uint     `json:"sum_other_doc_count"`
    Buckets                 []Bucket `json:"buckets"`
    Value                   uint64   `json:"value"`
}

type SearchResponseAggregations

type SearchResponseAggregations map[string]SearchResponseAggregation

type SearchResponseHit

type SearchResponseHit[T any] struct {
    Id      string `json:"_id"`
    Type    string `json:"_type"`
    Content T      `json:"_source"`
}

type SearchResponseHits

type SearchResponseHits[T any] struct {
    Total      SearchResponseHitsTotal
    SearchHits []SearchResponseHit[T] `json:"hits"`
}

type SearchResponseHitsTotal

type SearchResponseHitsTotal struct {
    Value    uint
    Relation string
}

type UpdateQueue

UpdateQueue is a queue for OpenSearch update requests.

type UpdateQueue struct {
    // contains filtered or unexported fields
}

func NewRequestQueue
func NewRequestQueue(openSearchProjectClient *opensearch.Client, updateMaxRetries int, updateRetryDelay time.Duration) *UpdateQueue

NewRequestQueue creates a new update queue.

openSearchProjectClient is the official OpenSearch client to wrap. Use NewOpenSearchProjectClient to create it. updateMaxRetries is the number of retries for update requests. updateRetryDelay is the delay between retries.

func (*UpdateQueue) Stop
func (q *UpdateQueue) Stop()

func (*UpdateQueue) Update
func (q *UpdateQueue) Update(indexName string, requestBody []byte) ([]byte, error)

Update queues and update for an index and returns the response body or an error

Is called from pkg/openSearch/open_search_client/client.go: func (c *Client) Update(indexName string, requestBody []byte) (responseBody []byte, err error) and tested in pkg/openSearch/open_search_client/client_test.go

indexName: The name of the index to update requestBody: The request body to send to the index

Returns: The response body or an error

Generated by gomarkdoc

License

Copyright (C) 2022-2023 [Greenbone AG][Greenbone AG]

Licensed under the GNU General Public License v3.0 or later.

Documentation

Overview

Package openSearchClient provides a client for OpenSearch designed to allow easy mocking in tests.

Example Usage: 

clientConfig, err := config.ReadOpensearchClientConfig()
if err != nil {
	return err
}

opensearchProjectClient, err := NewOpenSearchProjectClient(context.Background(), clientConfig)
if err != nil {
	return err
}

client := NewClient(opensearchProjectClient, 10, 1)

query := `{"query":{"bool":{"filter":[{"term":{"oid":{"value":"1.3.6.1.4.1.25623.1.0.117842"}}}]}}}`
responseBody, err := client.Search(indexName, []byte(query))
if err != nil {
	return err
}

searchResponse, err := UnmarshalSearchResponse[*Vulnerability](responseBody)
if err != nil {
	return err
}

For further usage examples see ./client_test.go.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func GetResponseError 

func GetResponseError(statusCode int, responseString []byte, indexName string) error

GetResponseError checks if a response from OpenSearch indicated success and returns an error if not.

func InitializeJson 

func InitializeJson(timeFormats []string)

func NewIndexFunction 

func NewIndexFunction(openSearchProjectClient *opensearch.Client) *indexFunction

func NewOpenSearchErrorWithStack 

func NewOpenSearchErrorWithStack(message string) error

func NewOpenSearchProjectClient 

func NewOpenSearchProjectClient(ctx context.Context, config config.OpensearchClientConfig) (*opensearch.Client, error)

NewOpenSearchProjectClient creates a new official OpenSearch client (package github.com/opensearch-project/opensearch-go) for usage NewClient. It returns an error if the client couldn't be created or the connection couldn't be established.

ctx is the context to use for the connection. config is the configuration for the client.

func NewOpenSearchResourceAlreadyExistsWithStack 

func NewOpenSearchResourceAlreadyExistsWithStack(message string) error

func NewOpenSearchResourceNotFoundWithStack 

func NewOpenSearchResourceNotFoundWithStack(message string) error

func SerializeDocumentsForBulkUpdate 

func SerializeDocumentsForBulkUpdate[T any](indexName string, documents []T) ([]byte, error)

SerializeDocumentsForBulkUpdate serializes documents for bulk update. Can be used in conjunction with BulkUpdate. It returns the serialized documents or an error in case something went wrong.

indexName is the name of the index to update. documents are the documents to update.

func StartOpensearchTestContainer 

func StartOpensearchTestContainer(ctx context.Context) (testcontainers.Container, config.OpensearchClientConfig, error)

StartOpensearchTestContainer starts a test container with opensearch and returns the container and the config for the opensearch client. It returns an error if the container couldn't be created or started.

ctx is the context to use for the container.

func Unmarshal 

func Unmarshal(data []byte, v any) error

Unmarshal unmarshalls data into v. It returns an error if the data is invalid.

func UnmarshalWithoutValidation 

func UnmarshalWithoutValidation(data []byte, v any) error

UnmarshalWithoutValidation unmarshalls data into v. It returns an error if the data can not be parsed.

Types

type Bucket 

type Bucket struct {
	Key         any    `json:"key"`
	KeyAsString string `json:"key_as_string"`
	DocCount    uint   `json:"doc_count"`
	Aggs        map[string]DynamicAggregation
}

func (*Bucket) UnmarshalJSON 

func (bucket *Bucket) UnmarshalJSON(bytes []byte) error

type BulkResponse 

type BulkResponse struct {
	Took     uint         `json:"took"`
	HasError bool         `json:"errors"`
	Errors   []IndexError `json:"items"`
}

BulkResponse bulk response

type Client added in v1.0.1 

type Client struct {
	// contains filtered or unexported fields
}

Client is a client for OpenSearch designed to allow easy mocking in tests. It is a wrapper around the official OpenSearch client github.com/opensearch-project/opensearch-go .

func NewClient 

func NewClient(openSearchProjectClient *opensearch.Client, updateMaxRetries int, updateRetryDelay time.Duration) *Client

NewClient creates a new OpenSearch client.

openSearchProjectClient is the official OpenSearch client to wrap. Use NewOpenSearchProjectClient to create it. updateMaxRetries is the number of retries for update requests. updateRetryDelay is the delay between retries.

func (*Client) AsyncDeleteByQuery added in v1.0.1 

func (c *Client) AsyncDeleteByQuery(indexName string, requestBody []byte) error

AsyncDeleteByQuery updates documents in the given index asynchronously. It does not wait for the update to finish before returning. It returns an error in case something went wrong.

indexName is the name of the index to delete from. requestBody is the request body to send to OpenSearch to identify the documents to be deleted.

func (*Client) BulkUpdate added in v1.0.1 

func (c *Client) BulkUpdate(indexName string, requestBody []byte) error

BulkUpdate performs a bulk update in the given index. It returns an error in case something went wrong.

indexName is the name of the index to update. requestBody is the request body to send to OpenSearch specifying the bulk update.

func (*Client) Close added in v1.0.1 

func (c *Client) Close()

Close stops the underlying UpdateQueue allowing a graceful shutdown.

func (*Client) DeleteByQuery added in v1.0.1 

func (c *Client) DeleteByQuery(indexName string, requestBody []byte) error

DeleteByQuery updates documents in the given index. It waits for the update to finish before returning. It returns an error in case something went wrong.

indexName is the name of the index to delete from. requestBody is the request body to send to OpenSearch to identify the documents to be deleted.

func (*Client) Search added in v1.0.1 

func (c *Client) Search(indexName string, requestBody []byte) (responseBody []byte, err error)

Search searches for documents in the given index.

indexName is the name of the index to search in. requestBody is the request body to send to OpenSearch. It returns the response body as or an error in case something went wrong.

func (*Client) Update added in v1.0.1 

func (c *Client) Update(indexName string, requestBody []byte) (responseBody []byte, err error)

Update updates documents in the given index using UpdateQueue (which is also part of this package). It does not wait for the update to finish before returning. It returns the response body as or an error in case something went wrong.

indexName is the name of the index to update. requestBody is the request body to send to OpenSearch specifying the update.

type CreatedResponse 

type CreatedResponse struct {
	Id     string `json:"_id"`
	Result string `json:"result"`
}

type DocumentError 

type DocumentError struct {
	IndexName  string            `json:"_index"`
	IndexType  string            `json:"_type"`
	DocumentId string            `json:"_id"`
	StatusCode uint              `json:"status"`
	Error      DocumentErrorType `json:"error"`
}

type DocumentErrorType 

type DocumentErrorType struct {
	Type   string `json:"type"`
	Reason string `json:"reason"`
}

type DynamicAggregation 

type DynamicAggregation struct {
	DocCountErrorUpperBound int                    `json:"doc_count_error_upper_bound"`
	SumOtherDocCount        uint                   `json:"sum_other_doc_count"`
	Buckets                 []Bucket               `json:"buckets"`
	Value                   any                    `json:"value"`
	ValueAsString           any                    `json:"value_as_string"`
	Hits                    DynamicAggregationHits `json:"hits"`
}

type DynamicAggregationHits 

type DynamicAggregationHits struct {
	Total      SearchResponseHitsTotal `json:"total"`
	SearchHits KeepJsonAsString        `json:"hits"`
}

type IndexError 

type IndexError struct {
	Index DocumentError `json:"index"`
}

type KeepJsonAsString 

type KeepJsonAsString []byte

func (*KeepJsonAsString) UnmarshalJSON 

func (k *KeepJsonAsString) UnmarshalJSON(data []byte) error

type OpenSearchError 

type OpenSearchError struct {
	Message string
}

OpenSearchError openSearch error

func NewOpenSearchError 

func NewOpenSearchError(message string) *OpenSearchError

func (*OpenSearchError) Error 

func (o *OpenSearchError) Error() string

type OpenSearchErrorResponse 

type OpenSearchErrorResponse struct {
	Error  OpenSearchErrors
	Status int
}

type OpenSearchErrors 

type OpenSearchErrors struct {
	Reasons []OpenSearchRootCause
	Type    string
	Reason  string
}

type OpenSearchResourceAlreadyExists 

type OpenSearchResourceAlreadyExists struct {
	Message string
}

OpenSearchResourceAlreadyExists openSearch resource already exists

func NewOpenSearchResourceAlreadyExists 

func NewOpenSearchResourceAlreadyExists(message string) *OpenSearchResourceAlreadyExists

func (*OpenSearchResourceAlreadyExists) Error 

func (o *OpenSearchResourceAlreadyExists) Error() string

type OpenSearchResourceNotFound 

type OpenSearchResourceNotFound struct {
	Message string
}

OpenSearchResourceNotFound openSearch resource already exists

func NewOpenSearchResourceNotFound 

func NewOpenSearchResourceNotFound(message string) *OpenSearchResourceNotFound

func (*OpenSearchResourceNotFound) Error 

func (o *OpenSearchResourceNotFound) Error() string

type OpenSearchRootCause 

type OpenSearchRootCause struct {
	Type   string
	Reason string
}

type OpensearchTestContainer 

type OpensearchTestContainer struct {
	testcontainers.Container
}

OpensearchTestContainer represents the opensearch container

type Request 

type Request struct {
	IndexName   string
	RequestBody []byte
	Response    chan Response // Use the new Response type
}

type Response 

type Response struct {
	Body []byte
	Err  error
}

type SearchResponse 

type SearchResponse[T any] struct {
	Took         uint                       `json:"took"`
	TimedOut     bool                       `json:"timed_out"`
	Hits         SearchResponseHits[T]      `json:"hits"`
	Aggregations SearchResponseAggregations `json:"aggregations"`
}

func UnmarshalSearchResponse 

func UnmarshalSearchResponse[T any](data []byte) (*SearchResponse[T], error)

func (SearchResponse[T]) GetResults 

func (s SearchResponse[T]) GetResults() []T

GetResults returns list of documents

func (SearchResponse[T]) GetSearchHits 

func (s SearchResponse[T]) GetSearchHits() []SearchResponseHit[T]

type SearchResponseAggregation 

type SearchResponseAggregation struct {
	DocCountErrorUpperBound int      `json:"doc_count_error_upper_bound"`
	SumOtherDocCount        uint     `json:"sum_other_doc_count"`
	Buckets                 []Bucket `json:"buckets"`
	Value                   uint64   `json:"value"`
}

type SearchResponseAggregations 

type SearchResponseAggregations map[string]SearchResponseAggregation

type SearchResponseHit 

type SearchResponseHit[T any] struct {
	Id      string `json:"_id"`
	Type    string `json:"_type"`
	Content T      `json:"_source"`
}

type SearchResponseHits 

type SearchResponseHits[T any] struct {
	Total      SearchResponseHitsTotal
	SearchHits []SearchResponseHit[T] `json:"hits"`
}

type SearchResponseHitsTotal 

type SearchResponseHitsTotal struct {
	Value    uint
	Relation string
}

type UpdateQueue added in v1.0.1 

type UpdateQueue struct {
	// contains filtered or unexported fields
}

UpdateQueue is a queue for OpenSearch update requests.

func NewRequestQueue 

func NewRequestQueue(openSearchProjectClient *opensearch.Client, updateMaxRetries int, updateRetryDelay time.Duration) *UpdateQueue

NewRequestQueue creates a new update queue.

openSearchProjectClient is the official OpenSearch client to wrap. Use NewOpenSearchProjectClient to create it. updateMaxRetries is the number of retries for update requests. updateRetryDelay is the delay between retries.

func (*UpdateQueue) Stop added in v1.0.1 

func (q *UpdateQueue) Stop()

func (*UpdateQueue) Update added in v1.0.1 

func (q *UpdateQueue) Update(indexName string, requestBody []byte) ([]byte, error)

Update queues and update for an index and returns the response body or an error

Is called from pkg/openSearch/open_search_client/client.go: func (c *Client) Update(indexName string, requestBody []byte) (responseBody []byte, err error) and tested in pkg/openSearch/open_search_client/client_test.go

indexName: The name of the index to update requestBody: The request body to send to the index

Returns: The response body or an error

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.