Allow cost-efficient operational analytics with Amazon OpenSearch Ingestion

As the size and complexity of microservices and distributed functions continues to broaden, clients are looking for steerage for constructing cost-efficient infrastructure supporting operational analytics use circumstances. Operational analytics is a well-liked use case with Amazon OpenSearch Service. A couple of of the defining traits of those use circumstances are ingesting a excessive quantity of time collection knowledge and a comparatively low quantity of querying, alerting, and operating analytics on ingested knowledge for real-time insights. Though OpenSearch Service is able to ingesting petabytes of information throughout storage tiers, you continue to should provision capability emigrate between scorching and heat tiers. This provides to the price of provisioned OpenSearch Service domains.

The time collection knowledge usually comprises logs or telemetry knowledge from numerous sources with totally different values and desires. That’s, logs from some sources should be out there in a scorching storage tier longer, whereas logs from different sources can tolerate a delay in querying and different necessities. Till now, clients have been constructing exterior ingestion techniques with the Amazon Kinesis household of companies, Amazon Easy Queue Service (Amazon SQS), AWS Lambda, customized code, and different comparable options. Though these options allow ingestion of operational knowledge with numerous necessities, they add to the price of ingestion.

Usually, operational analytics workloads use anomaly detection to help area operations. This assumes that the info is already current in OpenSearch Service and the price of ingestion is already borne.

With the addition of some current options of Amazon OpenSearch Ingestion, a totally managed serverless pipeline for OpenSearch Service, you’ll be able to successfully tackle every of those value factors and construct a cheap answer. On this publish, we define an answer that does the next:

• Makes use of conditional routing of Amazon OpenSearch Ingestion to separate logs with particular attributes and retailer these, for instance, in Amazon OpenSearch Service and archive all occasions in Amazon S3 to question with Amazon Athena
• Makes use of in-stream anomaly detection with OpenSearch Ingestion, thereby eradicating the associated fee related to compute wanted for anomaly detection after ingestion

On this publish, we use a VPC move logs use case to show the answer. The answer and sample introduced on this publish is equally relevant to bigger operational analytics and observability use circumstances.

Resolution overview

We use VPC move logs to seize IP site visitors and set off processing notifications to the OpenSearch Ingestion pipeline. The pipeline filters the info, routes the info, and detects anomalies. The uncooked knowledge will probably be saved in Amazon S3 for archival functions, then the pipeline will detect anomalies within the knowledge in near-real time utilizing the Random Minimize Forest (RCF) algorithm and ship these knowledge data to OpenSearch Service. The uncooked knowledge saved in Amazon S3 will be inexpensively retained for an prolonged time frame utilizing tiered storage and queried utilizing the Athena question engine, and likewise visualized utilizing Amazon QuickSight or different knowledge visualization companies. Though this walkthrough makes use of VPC move log knowledge, the identical sample applies to be used with AWS CloudTrail, Amazon CloudWatch, any log information in addition to any OpenTelemetry occasions, and customized producers.

The next is a diagram of the answer that we configure on this publish.

Within the following sections, we offer a walkthrough for configuring this answer.

The patterns and procedures introduced on this publish have been validated with the present model of OpenSearch Ingestion and the Knowledge Prepper open-source challenge model 2.4.

Conditions

Full the next prerequisite steps:

1. We will probably be utilizing a VPC for demonstration functions for producing knowledge. Arrange the VPC move logs to publish logs to an S3 bucket in textual content format. To optimize S3 storage prices, create a lifecycle configuration on the S3 bucket to transition the VPC move logs to totally different tiers or expire processed logs. Make an observation of the S3 bucket title you configured to make use of in later steps.
2. Arrange an OpenSearch Service area. Make an observation of the area URL. The area will be both public or VPC based mostly, which is the popular configuration.
3. Create an S3 bucket for storing archived occasions, and make a remark of S3 bucket title. Configure a resource-based coverage permitting OpenSearch Ingestion to archive logs and Athena to learn the logs.
4. Configure an AWS Id and Entry Administration (IAM) function or separate IAM roles permitting OpenSearch Ingestion to work together with Amazon SQS and Amazon S3. For directions, check with Configure the pipeline function.
5. Configure Athena or validate that Athena is configured in your account. For directions, check with Getting began.

Configure an SQS notification

VPC move logs will write knowledge in Amazon S3. After every file is written, Amazon S3 will ship an SQS notification to inform the OpenSearch Ingestion pipeline that the file is prepared for processing.

If the info is already saved in Amazon S3, you should utilize the S3 scan functionality for a one-time or scheduled loading of information by means of the OpenSearch Ingestion pipeline.

Use AWS CloudShell to situation the next instructions to create the SQS queues VpcFlowLogsNotifications and VpcFlowLogsNotifications-DLQ that we use for this walkthrough.

Create a dead-letter queue with the next code

export SQS_DLQ_URL=$(aws sqs create-queue --queue-name VpcFlowLogsNotifications-DLQ | jq -r '.QueueUrl')

echo $SQS_DLQ_URL 

export SQS_DLQ_ARN=$(aws sqs get-queue-attributes --queue-url $SQS_DLQ_URL --attribute-names QueueArn | jq -r '.Attributes.QueueArn') 

echo $SQS_DLQ_ARN

Create an SQS queue to obtain occasions from Amazon S3 with the next code:

export SQS_URL=$(aws sqs create-queue --queue-name VpcFlowLogsNotification --attributes '{
"RedrivePolicy": 
"{"deadLetterTargetArn":"'$SQS_DLQ_ARN'","maxReceiveCount":"2"}", 
"Coverage": 
  "{"Model":"2012-10-17","Assertion":[{"Effect":"Allow","Principal":{"Service":"s3.amazonaws.com"}, "Action":"SQS:SendMessage","Resource":"*"}]}" 
}' | jq -r '.QueueUrl')

echo $SQS_URL

To configure the S3 bucket to ship occasions to the SQS queue, use the next code (present the title of your S3 bucket used for storing VPC move logs):

aws s3api put-bucket-notification-configuration --bucket __BUCKET_NAME__ --notification-configuration '{
     "QueueConfigurations": [
         {
             "QueueArn": "'$SQS_URL'",
             "Events": [
                 "s3:ObjectCreated:*"
             ]
         }
     ]
}'

Create the OpenSearch Ingestion pipeline

Now that you’ve got configured Amazon SQS and the S3 bucket notifications, you’ll be able to configure the OpenSearch Ingestion pipeline.

1. On the OpenSearch Service console, select Pipelines below Ingestion within the navigation pane.
2. Select Create pipeline.

3. For Pipeline title, enter a reputation (for this publish, we use stream-analytics-pipeline).
4. For Pipeline configuration, enter the next code:

model: "2"
entry-pipeline:
  supply:
     s3:
       notification_type: sqs
       compression: gzip
       codec:
         newline:
       sqs:
         queue_url: "<sturdy>__SQS_QUEUE_URL__</sturdy>"
         visibility_timeout: 180s
       aws:
        area: "<sturdy>__REGION__</sturdy>"
        sts_role_arn: "<sturdy>__STS_ROLE_ARN__</sturdy>"
  
  processor:
  sink:
    - pipeline:
        title: "archive-pipeline"
    - pipeline:
        title: "data-processing-pipeline"

data-processing-pipeline:
    supply: 
        pipeline:
            title: "entry-pipeline"
    processor:
    - grok:
        tags_on_match_failure: [ "grok_match_failure" ]
        match:
          message: [ "%{VPC_FLOW_LOG}" ]
    route:
        - icmp_traffic: '/protocol == 1'
    sink:
        - pipeline:
            title : "icmp-pipeline"
            routes:
                - "icmp_traffic"
        - pipeline:
            title: "analytics-pipeline"
    

archive-pipeline:
  supply:
    pipeline:
      title: entry-pipeline
  processor:
  sink:
    - s3:
        aws:
          area: "<sturdy>__REGION__</sturdy>"
          sts_role_arn: "<sturdy>__STS_ROLE_ARN__</sturdy>"
        max_retries: 16
        bucket: "<sturdy>__AWS_S3_BUCKET_ARCHIVE__</sturdy>"
        object_key:
          path_prefix: "vpc-flow-logs-archive/yr=%{yyyy}/month=%{MM}/day=%{dd}/"
        threshold:
          maximum_size: 50mb
          event_collect_timeout: 300s
        codec:
          parquet:
            auto_schema: true
      
analytics-pipeline:
  supply:
    pipeline:
      title: "data-processing-pipeline"
  processor:
    - drop_events:
        drop_when: "hasTags("grok_match_failure") or "/log-status" == "NODATA""
    - date:
        from_time_received: true
        vacation spot: "@timestamp"
    - mixture:
        identification_keys: ["srcaddr", "dstaddr"]
        motion:
          tail_sampler:
            %: 20.0
            wait_period: "60s"
            situation: '/motion != "ACCEPT"'
    - anomaly_detector:
        identification_keys: ["srcaddr","dstaddr"]
        keys: ["bytes"]
        verbose: true
        mode:
          random_cut_forest:
  sink:
    - opensearch:
        hosts: [ "<strong>__AMAZON_OPENSEARCH_DOMAIN_URL__</strong>" ]
        index: "flow-logs-anomalies"
        aws:
          sts_role_arn: "<sturdy>__STS_ROLE_ARN__</sturdy>"
          area: "<sturdy>__REGION__</sturdy>"
          
icmp-pipeline:
  supply:
    pipeline:
      title: "data-processing-pipeline"
  processor:
  sink:
    - opensearch:
        hosts: [ "<strong>__AMAZON_OPENSEARCH_DOMAIN_URL__</strong>" ]
        index: "sensitive-icmp-traffic"
        aws:
          sts_role_arn: "<sturdy>__STS_ROLE_ARN__</sturdy>"
          area: "<sturdy>__REGION__</sturdy>"</code>

Exchange the variables within the previous code with assets in your account:

• • __SQS_QUEUE_URL__ – URL of Amazon SQS for Amazon S3 occasions
  • __STS_ROLE_ARN__ – AWS Safety Token Service (AWS STS) roles for assets to imagine
  • __AWS_S3_BUCKET_ARCHIVE__ – S3 bucket for archiving processed occasions
  • __AMAZON_OPENSEARCH_DOMAIN_URL__ – URL of OpenSearch Service area
  • __REGION__ – Area (for instance, us-east-1)

5. Within the Community settings part, specify your community entry. For this walkthrough, we’re utilizing VPC entry. We offered the VPC and personal subnet areas which have connectivity with the OpenSearch Service area and safety teams.
6. Go away the opposite settings with default values, and select Subsequent.
7. Overview the configuration adjustments and select Create pipeline.

It should take a couple of minutes for OpenSearch Service to provision the surroundings. Whereas the surroundings is being provisioned, we’ll stroll you thru the pipeline configuration. Entry-pipeline listens for SQS notifications about newly arrived information and triggers the studying of VPC move log compressed information:

…
entry-pipeline:
  supply:
     s3:
…

The pipeline branches into two sub-pipelines. The primary shops authentic messages for archival functions in Amazon S3 in read-optimized Parquet format; the opposite applies analytics routes occasions to the OpenSearch Service area for quick querying and alerting:

…
  sink:
    - pipeline:
        title: "archive-pipeline"
    - pipeline:
        title: "data-processing-pipeline"
… 

The pipeline archive-pipeline aggregates messages in 50 MB chunks or each 60 seconds and writes a Parquet file to Amazon S3 with the schema inferred from the message. Additionally, a prefix is added to assist with partitioning and question optimization when studying a set of information utilizing Athena.

…
sink:
    - s3:
…
        object_key:
          path_prefix: " vpc-flow-logs-archive/yr=%{yyyy}/month=%{MM}/day=%{dd}/"
        threshold:
          maximum_size: 50mb
          event_collect_timeout: 300s
        codec:
          parquet:
            auto_schema: true
…

Now that we now have reviewed the fundamentals, we concentrate on the pipeline that detects anomalies and sends solely high-value messages that deviate from the norm to OpenSearch Service. It additionally shops Web Management Message Protocols (ICMP) messages in OpenSearch Service.

We utilized a grok processor to parse the message utilizing a predefined regex for parsing VPC move logs, and likewise tagged all unparsable messages with the grok_match_failure tag, which we use to take away headers and different data that may’t be parsed:

…
    processor:
    - grok:
        tags_on_match_failure: [ "grok_match_failure" ]
        match:
          message: [ "%{VPC_FLOW_LOG}" ]
…

We then routed all messages with the protocol identifier 1 (ICMP) to icmp-pipeline and all messages to analytics-pipeline for anomaly detection:

…
   route:
        - icmp_traffic: '/protocol == 1'
    sink:
        - pipeline:
            title : "icmp-pipeline"
            routes:
                - "icmp_traffic"
        - pipeline:
            title: "analytics-pipeline"
…

Within the analytics pipeline, we dropped all data that may’t be parsed utilizing the hasTags methodology based mostly on the tag that we assigned on the time of parsing. We additionally eliminated all data that don’t comprises helpful knowledge for anomaly detection.

…
  - drop_events:
        drop_when: "hasTags("grok_match_failure") or "/log-status" == "NODATA""		
…

Then we utilized probabilistic sampling utilizing the tail_sampler processor for all accepted messages grouped by supply and vacation spot addresses and despatched these to the sink with all messages that weren’t accepted. This helps cut back the amount of messages throughout the chosen cardinality keys, with a concentrate on all messages that weren’t accepted, and retains a pattern illustration of messages that have been accepted.

…
mixture:
        identification_keys: ["srcaddr", "dstaddr"]
        motion:
          tail_sampler:
            %: 20.0
            wait_period: "60s"
            situation: '/motion != "ACCEPT"'
…

Then we used the anomaly detector processor to determine anomalies throughout the cardinality key pairs or supply and vacation spot addresses in our instance. The anomaly detector processor creates and trains RCF fashions for a hashed worth of keys, then makes use of these fashions to find out whether or not newly arriving messages have an anomaly based mostly on the educated knowledge. In our demonstration, we use bytes knowledge to detect anomalies:

…
anomaly_detector:
        identification_keys: ["srcaddr","dstaddr"]
        keys: ["bytes"]
        verbose: true
        mode:
          random_cut_forest:
…

We set verbose:true to instruct the detector to emit the message each time an anomaly is detected. Additionally, for this walkthrough, we used a non-default sample_size for coaching the mannequin.

When anomalies are detected, the anomaly detector returns a whole file and provides

"deviation_from_expected":worth,"grade":worth attributes that signify the deviation worth and severity of the anomaly. These values can be utilized to find out routing of such messages to OpenSearch Service, and use per-document monitoring capabilities in OpenSearch Service to alert on particular circumstances.

Presently, OpenSearch Ingestion creates as much as 5,000 distinct fashions based mostly on cardinality key values per compute unit. This restrict is noticed utilizing the anomaly_detector.RCFInstances.worth CloudWatch metric. It’s vital to pick a cardinality key-value pair to keep away from exceeding this constraint. As growth of the Knowledge Prepper open-source challenge and OpenSearch Ingestion continues, extra configuration choices will probably be added to supply higher flexibility round mannequin coaching and reminiscence administration.

The OpenSearch Ingestion pipeline exposes the anomaly_detector.cardinalityOverflow.rely metric by means of CloudWatch. This metric signifies plenty of key worth pairs that weren’t run by the anomaly detection processor throughout a time frame as the utmost variety of RCFInstances per compute unit was reached. To keep away from this constraint, plenty of compute items will be scaled out to offer further capability for internet hosting further situations of RCFInstances.

Within the final sink, the pipeline writes data with detected anomalies together with deviation_from_expected and grade attributes to the OpenSearch Service area:

…
sink:
    - opensearch:
        hosts: [ "__AMAZON_OPENSEARCH_DOMAIN_URL__" ]
        index: "anomalies"
…

As a result of solely anomaly data are being routed and written to the OpenSearch Service area, we’re in a position to considerably cut back the scale of our area and optimize the price of our pattern observability infrastructure.

One other sink was used for storing all ICMP data in a separate index within the OpenSearch Service area:

…
sink:
    - opensearch:
        hosts: [ "__AMAZON_OPENSEARCH_DOMAIN_URL__" ]
        index: " sensitive-icmp-traffic"
…

Question archived knowledge from Amazon S3 utilizing Athena

On this part, we evaluation the configuration of Athena for querying archived occasions knowledge saved in Amazon S3. Full the next steps:

1. Navigate to the Athena question editor and create a brand new database referred to as vpc-flow-logs-archive-database utilizing the next command:

CREATE DATABASE `vpc-flow-logs-archive`

1. 2. On the Database menu, select vpc-flow-logs-archive.
2. Within the question editor, enter the next command to create a desk (present the S3 bucket used for archiving processed occasions). For simplicity, for this walkthrough, we create a desk with out partitions.

CREATE EXTERNAL TABLE `vpc-flow-logs-data`(
  `message` string)
ROW FORMAT SERDE 
  'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 
STORED AS INPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
  's3://__AWS_S3_BUCKET_ARCHIVE__'
TBLPROPERTIES (
  'classification'='parquet', 
  'compressionType'='none'
)

4. Run the next question to validate which you could question the archived VPC move log knowledge:

SELECT * FROM "vpc-flow-logs-archive"."vpc-flow-logs-data" LIMIT 10;

As a result of archived knowledge is saved in its authentic format, it helps keep away from points associated to format conversion. Athena will question and show data within the authentic format. Nonetheless, it’s ideally suited to work together solely with a subset of columns or elements of the messages. You should use the regexp_split operate in Athena to separate the message within the columns and retrieve sure columns. Run the next question to see the supply and vacation spot tackle groupings from the VPC move log knowledge:

SELECT srcaddr, dstaddr FROM (
   SELECT regexp_split(message, ' ')[4] AS srcaddr, 
          regexp_split(message, ' ')[5] AS dstaddr, 
          regexp_split(message, ' ')[14] AS standing  FROM "vpc-flow-logs-archive"."vpc-flow-logs-data" 
) WHERE standing="OK" 
GROUP BY srcaddr, dstaddr 
ORDER BY srcaddr, dstaddr LIMIT 10;

This demonstrated which you could question all occasions utilizing Athena, the place archived knowledge in its authentic uncooked format is used for the evaluation. Athena is priced per knowledge scanned. As a result of the info is saved in a read-optimized format and partitioned, it permits additional cost-optimization round on-demand querying of archived streaming and observability knowledge.

Clear up

To keep away from incurring future fees, delete the next assets created as a part of this publish:

• OpenSearch Service area
• OpenSearch Ingestion pipeline
• SQS queues
• VPC move logs configuration
• All knowledge saved in Amazon S3

Conclusion

On this publish, we demonstrated tips on how to use OpenSearch Ingestion pipelines to construct a cost-optimized infrastructure for log analytics and observability occasions. We used routing, filtering, aggregation, and anomaly detection in an OpenSearch Ingestion pipeline, enabling you to downsize your OpenSearch Service area and create a cost-optimized observability infrastructure. For our instance, we used a knowledge pattern with 1.5 million occasions with a pipeline distilling to 1,300 occasions with predicted anomalies based mostly on supply and vacation spot IP pairs. This metric demonstrates that the pipeline recognized that lower than 0.1% of occasions have been of excessive significance, and routed these to OpenSearch Service for visualization and alerting wants. This interprets to decrease useful resource utilization in OpenSearch Service domains and might result in provisioning of smaller OpenSearch Service environments.

We encourage you to make use of OpenSearch Ingestion pipelines to create your purpose-built and cost-optimized observability infrastructure that makes use of OpenSearch Service for storing and alerting on high-value occasions. If in case you have feedback or suggestions, please go away them within the feedback part.

Concerning the Authors

Mikhail Vaynshteyn is a Options Architect with Amazon Internet Companies. Mikhail works with healthcare and life sciences clients to construct options that assist enhance sufferers’ outcomes. Mikhail makes a speciality of knowledge analytics companies.

Muthu Pitchaimani is a Search Specialist with Amazon OpenSearch Service. He builds large-scale search functions and options. Muthu is within the matters of networking and safety, and is predicated out of Austin, Texas.