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	Number of worker threads for the filter and output stages. Increase this value if events are backlogged or CPU is underutilized. Default: the number of CPU cores.
Pipeline Batch Size	Maximum events a worker collects before executing filters and outputs. Larger batches increase memory usage and may require a larger JVM heap size (LS_HEAP_SIZE). Default: 125.
Pipeline Batch Delay	Wait time in milliseconds before dispatching an undersized batch to a worker thread. Default: 50 ms.
Queue Type	Internal queuing model for event buffering between stages. Valid values: MEMORY: Default. Uses an in-memory queue. PERSISTED: A disk-based persistent queue.
Queue Max Bytes	The maximum amount of data 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	Maximum events written before a checkpoint is forced (persistent queue only). 0 means no limit. Default: 1024.

Warning

Saving and deploying the configuration triggers an instance restart. Proceed only when this will not affect 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.