Skip to content

README.md

Latest commit

 

History

History
1152 lines (764 loc) · 31.2 KB

README.md

File metadata and controls

1152 lines (764 loc) · 31.2 KB

openSearchClient

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

Package openSearchClient provides functionality for interacting with OpenSearch.

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 InitializeJson

func InitializeJson(timeFormats []string)

func InjectAuthenticationIntoClient

func InjectAuthenticationIntoClient(client *opensearchapi.Client, config config.OpensearchClientConfig, tokenReceiver TokenReceiver) error

InjectAuthenticationIntoClient is a function that sets up the authentication method for the OpenSearch client. client is the OpenSearch client to inject the authentication into. config is the configuration for the OpenSearch client. tokenReceiver is the token receiver for OpenID authentication and must implement the GetClientAccessToken function. It can be nil for basic authentication.

func NewOpenSearchProjectClient

func NewOpenSearchProjectClient(ctx context.Context, config config.OpensearchClientConfig, tokenReceiver TokenReceiver) (*opensearchapi.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 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 Authenticator

Authenticator is a struct that holds the necessary information for authenticating with OpenSearch.

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

func (*Authenticator) Perform

func (a *Authenticator) Perform(req *http.Request) (*http.Response, error)

Perform is a method that implements the opensearchtransport.Interface interface. It injects the authentication header into the request and then performs the request.

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 ByCreationDate

type ByCreationDate []IndexInfo

func (ByCreationDate) Len

func (a ByCreationDate) Len() int

func (ByCreationDate) Less

func (a ByCreationDate) Less(i, j int) bool

func (ByCreationDate) Swap

func (a ByCreationDate) Swap(i, j int)

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 *opensearchapi.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) CompositeAggStream

func (c *Client) CompositeAggStream(indexName string, requestBody []byte, ctx context.Context) (io.Reader, error)

func (*Client) Count

func (c *Client) Count(indexName string, requestBody []byte) (count int64, err error)

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 (*Client) Search

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) SearchStream

func (c *Client) SearchStream(indexName string, requestBody []byte, scrollTimeout time.Duration, ctx context.Context) (io.Reader, error)

func (*Client) SyncUpdate

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

SyncUpdate updates documents in the given index synchronously.

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 CountReq

type CountReq struct {
    Indices []string
    Body    io.Reader
    Header  http.Header
    Params  map[string]string
}

func (CountReq) GetRequest

func (r CountReq) GetRequest() (*http.Request, error)

GetRequest returns the *http.Request that gets executed by the client

type CountResp

type CountResp struct {
    Count  int64      `json:"count"`
    Shards ShardStats `json:"_shards"`
}

func (CountResp) Inspect

func (r CountResp) Inspect() opensearchapi.Inspect

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 IndexFunction

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

func NewIndexFunction

func NewIndexFunction(openSearchProjectClient *opensearchapi.Client) *IndexFunction

func (*IndexFunction) AliasExists

func (i *IndexFunction) AliasExists(aliasName string) (bool, error)

func (*IndexFunction) CreateIndex

func (i *IndexFunction) CreateIndex(indexName string, indexSchema []byte) error

CreateIndex creates an index

func (*IndexFunction) CreateOrPutAlias

func (i *IndexFunction) CreateOrPutAlias(aliasName string, indexNames ...string) error

func (*IndexFunction) DeleteAliasFromIndex

func (i *IndexFunction) DeleteAliasFromIndex(indexName string, aliasName string) error

func (*IndexFunction) DeleteIndex

func (i *IndexFunction) DeleteIndex(indexName string) error

func (*IndexFunction) ForceMerge

func (i *IndexFunction) ForceMerge(index string, maximumNumberOfSegments int) error

func (*IndexFunction) GetIndexSettings

func (i *IndexFunction) GetIndexSettings(index string) (map[string]interface{}, error)

func (*IndexFunction) GetIndexes

func (i *IndexFunction) GetIndexes(pattern string) ([]string, error)

func (*IndexFunction) GetIndexesForAlias

func (i *IndexFunction) GetIndexesForAlias(aliasName string) ([]string, error)

previously AliasPointsToIndex

func (*IndexFunction) IndexExists

func (i *IndexFunction) IndexExists(indexName string) (bool, error)

func (*IndexFunction) IndexHasAlias

func (i *IndexFunction) IndexHasAlias(indexNames []string, aliasNames []string) (bool, error)

func (*IndexFunction) RefreshIndex

func (i *IndexFunction) RefreshIndex(index string) error

func (*IndexFunction) RemoveIndexesFromAlias

func (i *IndexFunction) RemoveIndexesFromAlias(indexesToRemove []string, aliasName string) error

func (*IndexFunction) SetIndexSettings

func (i *IndexFunction) SetIndexSettings(index string, settingsBody io.Reader) error

type IndexInfo

type IndexInfo struct {
    Name         string
    CreationDate int // Store Unix timestamp
}

func ConvertToIndexInfo

func ConvertToIndexInfo(indices []opensearchapi.CatIndexResp) []IndexInfo

func SortIndexInfoByCreationDate

func SortIndexInfoByCreationDate(indexes []IndexInfo) []IndexInfo

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 OpenSearchHealth

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

func NewOpenSearchHealth

func NewOpenSearchHealth(openSearchProjectClient *opensearchapi.Client) *OpenSearchHealth

func (*OpenSearchHealth) GetDiskAllocationPercentage

func (h *OpenSearchHealth) GetDiskAllocationPercentage() (int, error)

func (*OpenSearchHealth) GetIndexesWithCreationDate

func (h *OpenSearchHealth) GetIndexesWithCreationDate(pattern string) ([]IndexInfo, error)

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 ShardStats

type ShardStats struct {
    Total      int `json:"total"`
    Successful int `json:"successful"`
    Skipped    int `json:"skipped"`
    Failed     int `json:"failed"`
}

type SyncUpdateClient

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

func NewSyncUpdateClient

func NewSyncUpdateClient(osClient *opensearchapi.Client, maxRetries int, retryDelay time.Duration) *SyncUpdateClient

func (*SyncUpdateClient) Update

func (s *SyncUpdateClient) Update(indexName string, requestBody []byte) ([]byte, error)

type TokenReceiver

TokenReceiver is an interface for receiving client access tokens.

type TokenReceiver interface {
    GetClientAccessToken(clientName, clientSecret string) (string, error)
    ClearClientAccessToken()
}

type UpdateQueue

UpdateQueue is a queue for OpenSearch update requests.

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

func NewRequestQueue

func NewRequestQueue(openSearchClient *opensearchapi.Client, updateMaxRetries int, updateRetryDelay time.Duration) *UpdateQueue

NewRequestQueue creates a new update queue.

openSearchClient is the official OpenSearch client. 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.