Migrate OSS data by using SMQ event notifications-Elasticsearch(ES)-阿里云帮助中心

When a file change in Object Storage Service (OSS) triggers an event notification in Simple Message Queue (SMQ) (formerly MNS), you can use the logstash-input-oss plugin in Alibaba Cloud Logstash to consume the OSS change event. You can then use the logstash-output-oss plugin to synchronize the data to another OSS bucket. This topic describes the configuration procedure.

Prerequisites

You have installed the logstash-input-oss and logstash-output-oss plugins.
You have activated OSS and prepared your data.
You have activated Simple Message Queue (SMQ) and ensured it is in the same region as your OSS service.

Note
In this example, the events are configured to trigger on object creation (specifically, ObjectCreated.PostObject and ObjectCreated.PutObject). When these events occur, the logstash-output-oss plugin synchronizes the new data to the destination OSS bucket.

Procedure

Step 1: Configure an event notification rule
Step 2: Configure a Logstash pipeline
Step 3: Verify the data synchronization result

Step 1: Configure an event notification rule

Log on to the OSS console.
In the left-side navigation pane, click Bucket List. Then, click the name of the target bucket.
Choose Data Processing > Event Notification.
In the Event Notification section, click Create Rule.
On the Create Rule panel, configure the event notification rule.

The event notification in this example is configured as follows. During actual use, replace the example values with your own business information. For more information about the parameters, see Set event notification rules. Set Rule Name to rules. For Event Type, select File Creation Operations, and then select the ObjectCreated.PutObject and ObjectCreated.PostObject checkboxes. In the Object Matching section, set Prefix Matching Method to Equal to (=) and Matching Rule to logstash/index. After you complete the OSS event configuration, click Next to configure the Subscription Endpoint. Select an existing topic, configure the queue information in the Subscription Endpoint section, and then click Confirm.
Click OK.
Optional: View the event notification rule.
1. Go to the Simple Message Queue (SMQ) console.
2. In the left-side navigation pane, click Event Notification.
3. In the top navigation bar, select a region to view the event notification rules for that region.

Step 2: Configure a Logstash pipeline

Go to the Logstash Clusters page.
Navigate to the target cluster.
1. In the top navigation bar, select the region where the cluster resides.
2. On the Logstash Clusters page, find the cluster and click its ID.
In the left-side navigation pane, click Pipelines.
Click Create Pipeline.

On the Create Pipeline Task page, enter a pipeline ID and configure the pipeline.

This topic uses the following sample pipeline configuration. Replace the parameter values to suit your environment.

input {
  oss {
    endpoint => "oss-cn-hangzhou-internal.aliyuncs.com"
    bucket => "zl-ossoutoss"
    access_key_id => "ALIBABA_CLOUD_ACCESS_KEY_ID"
    access_key_secret => "ALIBABA_CLOUD_ACCESS_KEY_SECRET"
    mns_settings => {
       endpoint => "18185036364****.mns.cn-hangzhou-internal.aliyuncs.com"
       queue => "zl-test"
    }
    codec => json {
      charset => "UTF-8"
    }
  }
}

output {
  oss {
    endpoint => "http://oss-cn-hangzhou-internal.aliyuncs.com"              
    bucket => "zl-log-output-test"                          
    access_key_id => "ALIBABA_CLOUD_ACCESS_KEY_ID"                 
    access_key_secret => "ALIBABA_CLOUD_ACCESS_KEY_SECRET"         
    prefix => "oss/database"                         
    recover => true                                      
    rotation_strategy => "size_and_time"                  
    time_rotate => 1                                     
    size_rotate => 1000
    temporary_directory => "/ssd/1/ls-cn-0pp1cwec****/logstash/data/22"                            
    encoding => "gzip"                                 
    additional_oss_settings => {
      max_connections_to_oss => 1024                      
      secure_connection_enabled => false                  
    }

  }
}

Important

For the input block, the OSS endpoint must be an internal endpoint and must not start with http://.
Use time_rotate and size_rotate to trigger a data upload when data in the temporary directory is stored for 1 minute or reaches 1000 Bytes in size.
For more information about input parameters, see Parameters for the logstash-input-oss plugin. For more information about output parameters, see Parameters for the logstash-output-oss plugin.

Click Next step and configure the pipeline parameters.

Parameter	Description
Pipeline Workers	The number of worker threads to run the filter and output stages of the pipeline in parallel. If you have an event backlog or your CPU is underutilized, consider increasing this value to improve performance. Default value: the number of CPU cores in the instance.
Pipeline Batch Size	The maximum number of events a worker thread collects from inputs before executing filters and outputs. A larger batch size may lead to higher memory overhead. To use a larger batch size effectively, you may need to increase the JVM heap size by setting the LS_HEAP_SIZE variable. Default value: 125.
Pipeline Batch Delay	The duration in milliseconds to wait for each event before dispatching a small batch to a pipeline worker thread. Default value: 50 ms.
Queue Type	The internal queuing model for event buffering. Valid values: MEMORY: The default value. This specifies a traditional in-memory queue. PERSISTED: A disk-based persistent queue.
Queue Max Bytes	The maximum amount of data that the queue can store, in `MB`. The value must be an integer from `1` to `2⁵³-1`. Default value: `1024`. Note Make sure that this value is less than your total disk capacity.
Queue Checkpoint Writes	When the persistent queue is enabled, this is the maximum number of events that can be written before a checkpoint is forced. A value of 0 indicates no limit. Default value: 1024.

Warning

After configuration, you must save and deploy the settings for them to take effect. This action triggers an instance restart. Proceed only if this restart will not impact your business.

Click Save or Save and Deploy.
- Save: Saves the pipeline configuration but does not apply it. After saving, you are returned to the Pipelines page. In the Pipelines section, you can click Deploy Now in the Actions column to restart the instance and apply the configuration.
- Save and Deploy: Saves and deploys the configuration, restarting the instance to apply the changes.

Step 3: Verify data synchronization

Go to the OSS console.
Upload a new file to the directory that is monitored by SMQ.
Go to the destination OSS bucket and verify that the data is synchronized.

Important
The OSS plugins do not synchronize data on a per-file or per-directory basis. Instead, they read and write data from the source file event by event. Based on the rotation rule, the logstash-output-oss plugin first stores data in a temporary local file and uploads the data only when a configured time or size threshold is reached. Therefore, this process does not guarantee that data from a single source file will be saved to a single destination file. For example, data from multiple source files may be consolidated into one destination file.