Oracle Cloud Infrastructure Documentation

Creating a Connector with a Streaming Source

Create a connector in Connector Hub to transfer stream data from the Streaming service to a target service.

For more information about the Streaming service, see Streaming.

A connector that's defined with a Streaming source and (optional) function task supports the following targets: Functions, Log Analytics, Object Storage, and Streaming. The Notifications target is supported only when no function task is used.

For an example of the Connector Hub workflow, see Overview of Connector Hub. For an example of a connector that uses Monitoring as the source, see Scenario: Sending Metrics to Object Storage.

Note

  • For stream input schema, see Message Reference.
  • For a Notifications target with Streaming source, all messages are sent as raw JSON blobs.

Retention Period: Streaming Source

The retention period for the Streaming source in Connector Hub is customer-defined. See Limits on Streaming Resources. For more information about delivery, see Delivery Details.

Together with the retention period, the Streaming source's read position determines where in the stream to start moving data. You specify the read position when you specify the source connection.

Private Endpoint Prerequisites for Streams

Note

It's not possible to retrofit an existing connector to use a stream with a private endpoint. If the stream pool selected for the stream source or target is public, it can't be changed to private, and the connector can't be updated to reference a private stream. To change the source or target to use a different private stream, or to use a source or target other than Streaming, re-create the connector with the source and target that you want. An example of the need for a different private stream is a stream that was moved to a different stream pool. In that case, re-create the connector using the moved stream. Ensure that you deactivate or delete the old connector with the stream source or target that you don't want any more.

Private endpoint configuration is supported for source and target streams. Following are prerequisites for accessing streams that use private endpoints.

  • Ensure that you're granted permissions according to the following policies:
    allow group <group-name> to manage virtual-network-family in compartment id <compartment-ocid>
  where any {request.operation='CreatePrivateEndpoint',
    request.operation='UpdatePrivateEndpoint',
    request.operation='DeletePrivateEndpoint',
    request.operation='ChangePrivateEndpointCompartment',
    request.operation='EnableReverseConnection',
    request.operation='ModifyReverseConnection',
    request.operation='DisableReverseConnection' 
}
    allow group <group-name> to read stream-family in compartment id <compartment-ocid>
  • Best security practice is to locate a private stream and its connector in the same tenancy.

After creating the connector, configure ingress and egress rules to let the connector access that private endpoint. See Private Endpoints: Configure Ingress and Egress Rules.

  • On the Connectors list page, select Create connector. If you need help finding the list page, see Listing Connectors.

    Note

    When your configuration requires policies, default policies are offered. To accept a default policy, select the provided Create policy link.

    Default policies are offered for any authorization required for this connector to access source, task, and target services.

    You can get this authorization through these default policies or through group-based policies. The default policies are offered whenever you use the Console to create or edit a connector. The only exception is when the exact policy already exists in IAM, in which case the default policy isn't offered. For more information about this authorization requirement, see Authentication and Authorization.

    • If you don't have permissions to accept default policies, contact your administrator.
    • Automatically created policies remain when connectors are deleted. As a best practice, delete associated policies when deleting the connector.

    To review a newly created policy, select the associated view link.

    1. Basic Connector Information

    Enter identifying information and optional logs and tags.

    • Connector name: Enter a user-friendly name for the new connector and an optional description. Avoid entering confidential information.
    • Select a compartment: Select the compartment to store the new connector in.
    • Description (Optional): Enter a description for the connector.
    • Enable logs (Optional): Select to enable service logs for the new connector, and provide the following information.
      • Category (default value: Run Log)
      • Compartment: Select the compartment that you want for storing the service logs for the connector.
      • Log group: Select the log group (by compartment) that you want for storing the service logs. To create a new log group, select Create new group and then enter a name.
      • Log name (Optional): Enter a name for the log.
      • Show advanced options:
        • Enable legacy archival logs (Optional): Legacy log archival automatically creates a bucket in your compartment and archives a copy of the log.
        • Log retention (Optional): Specify how long to keep the service logs (default: 30 days).
      • Tags (Optional): To add tags to the connector, select the option that you see.
        • Tags
        • Add Tags (under Show Advanced Options)

        If you have permissions to create a resource, then you also have permissions to apply free-form tags to that resource. To apply a defined tag, you must have permissions to use the tag namespace. For more information about tagging, see Resource Tags. If you're not sure whether to apply tags, skip this option or ask an administrator. You can apply tags later.

    Select Next.

    2. Configure Connector Source

    Specify the source stream for the connector to transfer data from.

    • Select source: Select Streaming.
    • Configure source: Select the source stream.
      Note

      Private endpoint configuration is supported. To use a private endpoint, see Private Endpoint Prerequisites for Streams.
      • Compartment: Select the compartment that contains the stream that you want.
      • Stream pool: Select the stream pool that contains the stream that you want.
        Note

        To select a stream pool and stream for a connector, you must have authorization to read the stream pool and stream. See IAM Policies (Securing Connector Hub).
      • Stream: Select the name of the stream that you want to receive data from.
      • Read position: Specify the cursor position from which to start reading the stream.
        • Latest: Starts reading messages published after creating the connector.
          • If the first run of a new connector with this configuration is successful, then it moves data from the connector's creation time. If the first run fails (such as with missing policies), then after resolution the connector either moves data from the connector's creation time or, if the creation time is outside the retention period, the oldest available data in the stream. For example, consider a connector created at 10 a.m. for a stream with a two-hour retention period. If failed runs are resolved at 11 a.m., then the connector moves data from 10 a.m. If failed runs are resolved at 1 p.m., then the connector moves the oldest available data in the stream.
          • Later runs move data from the next position in the stream. If a later run fails, then after resolution the connector moves data from the next position in the stream or the oldest available data in the stream, depending on the stream's retention period.
        • Trim Horizon: Starts reading from the oldest available message in the stream.
          • If the first run of a new connector with this configuration is successful, then it moves data from the oldest available data in the stream. If the first run fails (such as with missing policies), then after resolution the connector moves the oldest available data in the stream, regardless of the stream's retention period.
          • Later runs move data from the next position in the stream. If a later run fails, then after resolution the connector moves data from the next position in the stream or the oldest available data in the stream, depending on the stream's retention period.

    Select Next.

    3. Configure Connector Task

    Optionally configure a function task to process data from the source using the Functions service.

    • Compartment: Select the compartment that contains the function that you want.
    • Function application: Select the name of the function application that includes the function you want.
    • Function: Select the name of the function that you want to use to process the data received from the source.

      For use by the connector as a task, the function must be configured to return one of the following responses:

      • List of JSON entries (must set the response header Content-Type=application/json)
      • Single JSON entry (must set the response header Content-Type=application/json)
      • Single binary object (must set the response header Content-Type=application/octet-stream)
    • Show additional options (Optional)
      • Batch options (Optional): Specify limits for each batch of data sent to the function.
        • Use automatic settings
        • Use manual settings: Provide values for batch size limit (KBs) and batch time limit (seconds).

    Considerations for function tasks:

    • Connector Hub doesn't parse the output of the function task. The output of the function task is written as-is to the target. For example, when using a Notifications target with a function task, all messages are sent as raw JSON blobs.
    • Functions are invoked synchronously with 6 MB of data per invocation. If data exceeds 6 MB, then the connector invokes the function again to move the data that's over the limit. Such invocations are handled sequentially.
    • Functions can execute for up to five minutes. See Delivery Details.
    • Function tasks are limited to scalar functions.

    Select Next.

    4. Configure Connector Target

    Specify the target service to send the data to.

    • Target: Select the service that you want to transfer the data to.
      • Functions: Send stream data to a function.
      • Logging Analytics: Send stream data to a log group.
      • Notifications: Send stream data to a topic. Notifications is supported only when no function task is defined.
      • Object Storage: Send stream data to a bucket.
      • Streaming: Send stream data to a stream.
    • Configure target: Enter values for the selected target.
      Functions
      • Compartment: Select the compartment that contains the function that you want.
      • Function application: Select the name of the function application that contains the function that you want.
      • Function: Select the name of the function that you want to send the data to.
      • Show additional options (Optional): Select this link and specify limits for each batch of data sent to the function. To use manual settings, provide values for batch size limit (either KBs or number of messages) and batch time limit (seconds).

        For example, limit batch size by selecting either 5,000 kilobytes or 10 messages. An example batch time limit is 5 seconds.

      Considerations for Functions targets:

      • The connector flushes source data as a JSON list in batches. Maximum batch, or payload, size is 6 MB.
      • Functions are invoked synchronously with 6 MB of data per invocation. If data exceeds 6 MB, then the connector invokes the function again to move the data that's over the limit. Such invocations are handled sequentially.
      • Functions can execute for up to five minutes. See Delivery Details.
      • Don't return data from Functions targets to connectors. Connector Hub doesn't read data returned from Functions targets.
      Logging Analytics
      • Compartment: Select the compartment that contains the log group that you want.
      • Log group: Select the log group that you want.
      • Log source identifier (for Streaming source only): Select the log source.
      Notifications
      • Compartment: Select the compartment that contains the topic that you want.
      • Topic: Select the name of the topic that you want to send the data to.
      • Message format: Select the option that you want:
        Note

        Message format options are available for connectors with Logging source only. These options aren't available for connectors with function tasks. When Message format options aren't available, messages are sent as raw JSON blobs.
        • Send formatted messages: Simplified, user-friendly layout.

          To view supported subscription protocols and message types for formatted messages, see Friendly Formatting.

        • Send raw messages: Raw JSON blob.

      Considerations for Notifications targets:

      • The maximum message size for the Notifications target is 128 KB. Any message that exceeds the maximum size is dropped.
      • SMS messages exhibit unexpected results for certain connector configurations. This issue is limited to topics that contain SMS subscriptions for the indicated connector configurations. For more information, see Multiple SMS messages for a single notification.
      Object Storage
      • Compartment: Select the compartment that contains the bucket that you want.
      • Bucket: Select the name of the bucket that you want to send the data to.
      • Object name prefix: Optionally enter a prefix value.
      • Show batch options: Select this link and optionally enter values for batch size (in MBs) and batch time (in milliseconds). Fields are labeled Batch size and Batch time.

      Considerations for Object Storage targets:

      • Batch rollover details:

        • Batch rollover size: 100 MB
        • Batch rollover time: 7 minutes

      • Files saved to Object Storage are compressed using gzip.

      • Format of data moved from a Monitoring source: Objects. The connector partitions source data from Monitoring by metric namespace and writes the data for each group (namespace) to an object. Each object name includes the following elements.

        <object_name_prefix>/<service_connector_ocid>/<metric_compartment_ocid>/<metric_namespace>/<data_start_timestamp>_<data_end_timestamp>.<sequence_number>.<file_type>.gz

        Within an object, each set of data points is appended to a new line.

      Streaming
      Note

      To select a stream pool and stream for a connector, you must have authorization to read the stream pool and stream. See IAM Policies (Securing Connector Hub). Private endpoint configuration is supported. To use a private endpoint, see Private Endpoint Prerequisites for Streams.
      • Compartment: Select the compartment that contains the stream that you want.
      • Stream: Select the name of the stream that you want to send the data to.

      Considerations for Streaming targets:

      • Format of data moved from a Monitoring source: Each object is written as a separate message.

    Select Next.

    Connector Preview

    Review the connector configuration and then select Create.

    The creation process begins, and its progress is displayed. On completion, the connector's details page opens.

  • Use the oci sch service-connector create command and required parameters to create a connector with a Streaming source:

    oci sch service-connector create --display-name "<display_name>" --compartment-id <compartment_OCID> --source [<stream_source_in_JSON>] --target [<target_in_JSON>]

    For a complete list of parameters and values for CLI commands, see the CLI Command Reference.

  • Run the CreateServiceConnector operation to create a connector.

    To create a connector with a Streaming source, populate source in the request (CreateServiceConnectorDetails) with streaming details. For an example, see StreamingSourceDetails.

Private Endpoints: Configure Ingress and Egress Rules

If you created a connector with a stream that uses a private endpoint, then configure ingress and egress rules to let the connector access that private endpoint.

  • Get IP addresses for configuration:
  • Allow primary traffic: Add an ingress rule to the NSG or security list with the following configuration.

    Console :

    • Stateless: No
    • Source: rceTrafficIpAddress (Example: 10.0.0.249/32)
    • IP Protocol: TCP
    • Source Port Range: All
    • Destination Port Range: All

    API - example rule in request: 

    {
  "protocol": "6",
  "source": "10.0.0.249/32"
}
  • Add an egress rule to the security list with the following configuration.

    Console :

    • Stateless: No
    • Source: streamPoolIpAddress (Example: 10.0.0.20/32)
    • IP Protocol: TCP
    • Source Port Range: All
    • Destination Port Range: All

    API - example rule in request: 

    {
  "destination": "10.0.0.20/32",
  "protocol": "6"
}
Note

It's not possible to retrofit an existing connector to use a stream with a private endpoint. If the stream pool selected for the stream source or target is public, it can't be changed to private, and the connector can't be updated to reference a private stream. To change the source or target to use a different private stream, or to use a source or target other than Streaming, re-create the connector with the source and target that you want. An example of the need for a different private stream is a stream that was moved to a different stream pool. In that case, re-create the connector using the moved stream. Ensure that you deactivate or delete the old connector with the stream source or target that you don't want any more.

Confirm That the New Connector Moves Data

After you create the connector, confirm that it's moving data.

  • Enable logs for the connector to get details on data flow.
  • Check for expected results at the target service.

Confirming that data is moved helps you avoid automatic deactivation, which happens when a connector fails for a long time.

Private streams: It's not possible to retrofit an existing connector to use a stream with a private endpoint. If the stream pool selected for the stream source or target is public, it can't be changed to private, and the connector can't be updated to reference a private stream. To change the source or target to use a different private stream, or to use a source or target other than Streaming, re-create the connector with the source and target that you want. An example of the need for a different private stream is a stream that was moved to a different stream pool. In that case, re-create the connector using the moved stream. Ensure that you deactivate or delete the old connector with the stream source or target that you don't want any more.