注意：

本教程需要访问 Oracle Cloud。要注册免费账户，请参阅开始使用 Oracle Cloud Infrastructure 免费套餐。
它对 Oracle Cloud Infrastructure 身份证明、租户和区间使用示例值。完成实验室后，请使用特定于云环境的那些值替换这些值。

从 Oracle Access Governance Event Data Publisher 读取发布到 OCI 流的消息

简介

事件数据发布是一个将一次性数据导出并持续向外部系统（例如 Oracle Cloud Infrastructure (OCI) 云账户）发布持续数据事件的流程。借助 Oracle Access Governance，您可以灵活地将一次性数据事件导出到云租户，并持续发布数据事件，例如身份、身份集合、策略、资源、资源访问等。您可以使用此数据获取洞察、存储数据以实现合规性，或者分析访问管理和治理数据。

事件是指在创建、修改或删除 Oracle Access Governance 组件（例如身份、策略、资源等）时发生的任何数据状态更改。通过使用事件数据发布器，管理员可以完全控制访问和身份数据，并可以使用它来自动记录事件并简化合规性报告。

数据事件发布流使用 OCI 存储桶进行一次性导出，并根据文件大小将后续更新发布到 OCI 流或 OCI 存储桶。

数据事件发布流

在本教程中，我们将介绍如何通过 Oracle Access Governance 捕获 OCI 中的用户更新并将其发布到 OCI 流，以及如何使用 OCI Cloud Shell 脚本对这些流消息进行使用和解码。

注：事件数据发布程序支持导出身份、身份集合、策略和资源，但是，本教程重点介绍如何捕获身份的事件。

Shell 脚本概述

此 shell 脚本的用途是检索在 OCI 流中发布的消息并对其进行解码。这些消息的编码方式使其无法以原始格式读取。通过运行脚本，您可以以清晰、人可读的格式查看消息，确保即使关键信息从 OCI 控制台消失后仍可访问。

具体而言，该脚本会创建一个游标来标记从指定 OCI 流读取消息的起点。游标基于指定的偏移量和通过生成测试流消息提取其值的分区。使用游标，脚本从 OCI 流中提取消息。最终，脚本处理并解码每个检索到的消息。

目标读者

Oracle Access Governance 和 Oracle Cloud Infrastructure Identity and Access Management (OCI IAM) 管理员。

目标

检索 OCI 流的密钥参数。
从控制台修改一个或多个 OCI 用户的几个属性。
在 Oracle Access Governance for OCI 中执行数据加载。
配置 OCI Cloud Shell 环境以运行脚本。
通过运行脚本读取流消息。

先决条件

具有管理权限的 Oracle Access Governance 实例。有关更多信息，请参阅设置服务实例和关于应用程序角色。
与 Oracle Access Governance 集成的 OCI 租户。有关详细信息，请参阅与 Oracle Cloud Infrastructure (OCI) 身份和访问管理 (IAM) 集成。
从 Oracle Access Governance 到 OCI 配置的事件数据发布程序。有关更多信息，请参见 Configure Event Data Publisher in Oracle Access Governance 。

任务 1：检索 OCI 流的关键参数

在此任务中，我们将登录到 OCI 控制台以检索流的密钥参数。

登录到 OCI 控制台，导航到 Analytics & AI 、 Messaging ，然后单击 Streaming 。
确保选择流所在的区间，然后单击流以查看详细信息。
记下流 OCID 和消息端点的值。
单击生成测试消息，将示例消息输入为数据，然后单击生成。它应返回 Success（成功）消息。
单击取消以关闭测试流窗口。
单击加载消息可检查测试消息。
在 Recent Messages 部分中，记下 Partition 和 Offset 值。

任务 2：更新 OCI 中的用户属性

在本任务中，我们将更改 OCI 中的用户属性。

转到 OCI 控制台，导航到身份和安全，然后单击域。
选择您的域并单击用户。单击要更新的用户，然后单击编辑用户。
更新国家/地区、标题、部门、成本中心、用户类型、员工编号等值。
单击保存更改。

注：（可选）您可以根据需要再更新几个用户。

任务 3：在 Oracle Access Governance 中执行数据加载

在此任务中，我们将在 Oracle Access Governance 中执行数据加载，以同步 OCI 更改。

登录到 Oracle Access Governance，导航到 Service Administration（服务管理）和 Orchestrated Systems（编排的系统）。
找到 OCI 的编排系统，单击三点（和管理集成）图标，然后选择管理集成。
单击立即加载数据并等待数据加载完成。

任务 4：配置 OCI Cloud Shell 以运行脚本来从流捕获消息

在此任务中，我们将配置 OCI Cloud Shell，并使用提供的脚本对 Oracle Access Governance 发布的消息进行解码。

转到 OCI 控制台，从右上角单击云 Shell ，然后等待云 Shell 窗口初始化。
初始化以下参数。
```
export STREAM_OCID=<STREAM_OCID_VALUE>
export ENDPOINT=<MESSAGES_ENDPOINT>
export CURSOR_TYPE="AFTER_OFFSET"
export PARTITION=<PARTITION>
export OFFSET=<OFFSET>
```
- 将流的 STREAM_OCID、ENDPOINT、PARTITION 和 OFFSET 参数替换为任务 1 中捕获的值。
- CURSOR_TYPE 设置为 AFTER_OFFSET，以包括在指定偏移值之后生成的所有消息进行读取。不要修改此值。
  
  注：可以从测试消息中获取偏移的示例值，但是您应该选择一个值，之后将捕获消息。这是 CURSOR_TYPE 的作用，就像 AFTER_OFFSET 一样。

创建 shell 脚本文件并使其可执行。

touch ag-streaming.sh
chmod u+x ag-streaming.sh

使用 vi 编辑器打开脚本。
```
vi ag-streaming.sh
```

按 i 进入 Insert Mode 。粘贴以下内容，然后按 Esc 和 ：wq！ 保存更改。

#!/bin/bash

# Validate required environment variables

required_vars=("STREAM_OCID" "ENDPOINT" "CURSOR_TYPE" "PARTITION" "OFFSET")

for var in "${required_vars[@]}"; do
   if [ -z "${!var}" ]; then
      echo "Error: Environment variable $var is not set."
      exit 1
   fi
done

# Create a cursor for the OCI Stream
# echo "Creating a cursor for the OCI Stream..."
oci streaming stream cursor create-cursor \
   --partition "$PARTITION" \
   --stream-id "$STREAM_OCID" \
   --type "$CURSOR_TYPE" \
   --offset "$OFFSET" \
   --endpoint "$ENDPOINT" > cursor.json

# Extract cursor value
cursor=$(jq -r '.data.value' cursor.json)
if [ -z "$cursor" ]; then
   echo "Error: Failed to retrieve cursor value."
   exit 1
fi

# Read messages from the OCI Stream
echo "Reading messages from the OCI Stream..."
messages=$(oci streaming stream message get \
   --stream-id "$STREAM_OCID" \
   --cursor "$cursor" \
   --endpoint "$ENDPOINT")

# Check if messages were retrieved
if [ -z "$messages" ] || ! echo "$messages" | jq -e '.data[] | select(.key != null)' > /dev/null 2>&1; then
   echo "No valid messages found."
   exit 1
fi

# Extract and decode the data from messages
echo "Processing messages..."

# Extract and count the number of messages
message_count=$(echo "$messages" | jq -r '.data[] | select(.key != null) | .value' | wc -l)

echo "$messages" | jq -r '.data[] | select(.key != null) | .value' | while read -r value; do
   if [ -n "$value" ]; then
      counter=$((counter + 1))
      echo
      echo "Decoding message $counter of $message_count..."

      # Base64 decode the message
      decoded_message=$(echo "$value" | base64 --decode 2>/dev/null || echo "Error decoding message")
      final_decoded_message=$(echo "$decoded_message" | base64 --decode 2>/dev/null || echo "Error decoding message further")

      # Print the decoded message
      echo "Decoded Message : $final_decoded_message"
   fi
done

运行脚本并将输出消息重定向到文件（以 messages_dump.txt 为例）。或者，您可以运行脚本在控制台上显示消息。
```
./ag-streaming.sh > messages_dump.txt
```
从 OCI Cloud Shell 下载 messages_dump.txt 文件，以检查解码的消息以了解以前进行的用户更新。

在本教程中，我们学习了如何从 Oracle Access Governance 中的事件数据发布者发布的 OCI 流中读取消息。我们了解了如何使用 OCI Cloud Shell 利用 OCI 命令行界面 (CLI) 根据指定的偏移读取消息。这些消息包含与 OCI 用户、组、资源和策略关联相关的更改，因此从审计、合规性和报告角度来看非常重要。

后续步骤

完成这些教程后，您可以分析从 Oracle Access Governance 发布到 OCI 流的数据事件消息的内容。您可以根据需要设置进一步流程来分析事件消息，以满足审计和合规性要求。您还可以将数据从流迁移到自治数据仓库，以执行高级分析和可视化。还可以将数据移至 OCI Object Storage 进行长期存储，或运行 Hadoop/Spark 作业。

确认

作者 — Anuj Tripathi（北美解决方案工程首席云架构师）