使用 OpenTelemetry 和 Jaeger 对 MinIO 进行分布式追踪

$ mkdir -p /minio/data

$ docker run -d \
  -p 9000:9000 \
  -p 9001:9001 \
  --name minio \
  -v /minio/data:/data \
  -e "MINIO_ROOT_USER=minio" \
  -e "MINIO_ROOT_PASSWORD=minioadmin" \
  quay.io/minio/minio server /data --console-address ":9001"

$ pip install minio

from minio import Minio

# Convenient dict for basic config
config = {
  "dest_bucket":    "processed", # This will be auto created
  "minio_endpoint": "localhost:9000",
  "minio_username": "minio",
  "minio_password": "minioadmin",
}

# Initialize MinIO client
minio_client = Minio(config["minio_endpoint"],
              secure=False,
              access_key=config["minio_username"],
              secret_key=config["minio_password"]
              )

# Create destination bucket if it does not exist
if not minio_client.bucket_exists(config["dest_bucket"]):
  minio_client.make_bucket(config["dest_bucket"])
  print("Destination Bucket '%s' has been created" % (config["dest_bucket"]))

# Create a test object
file_path = "test_object.txt"
f = open(file_path, "w")
f.write("created test object")
f.close()

# Put an object inside the bucket
minio_client.fput_object(config["dest_bucket"], file_path, file_path)

# Get the object from the bucket
minio_client.fget_object(config["dest_bucket"], file_path, config["dest_bucket"] + "/" + file_path)

# Get list of objects
for obj in minio_client.list_objects(config["dest_bucket"]):
  print(obj)
  print("Some objects here")

from minio import Minio

# Convenient dict for basic config
config = {
  "dest_bucket":    "processed", # This will be auto created
  "minio_endpoint": "localhost:9000",
  "minio_username": "minio",
  "minio_password": "minioadmin",
}

# Initialize MinIO client
minio_client = Minio(config["minio_endpoint"],
              secure=False,
              access_key=config["minio_username"],
              secret_key=config["minio_password"]
              )

# 如果目标桶不存在，则创建
if not minio_client.bucket_exists(config["dest_bucket"]):
  minio_client.make_bucket(config["dest_bucket"])
  print("Destination Bucket '%s' has been created" % (config["dest_bucket"]))

# 创建一个测试对象
file_path = "test_object.txt"
f = open(file_path, "w")
f.write("created test object")
f.close()

minio_client.fput_object(config["dest_bucket"], file_path, file_path)

minio_client.fget_object(config["dest_bucket"], file_path, config["dest_bucket"] + "/" + file_path)

for obj in minio_client.list_objects(config["dest_bucket"]):
  print(obj)
  print("这里有一些对象")

<minio.datatypes.Object object at 0x109b1d3d0>
这里有一些对象

$ python minio_app.py

minio_client.remove_object(config["dest_bucket"], file_path)

for obj in minio_client.list_objects(config["dest_bucket"]):
  print(obj)
  print("No objects here")

if minio_client.bucket_exists(config["dest_bucket"]):
  minio_client.remove_bucket(config["dest_bucket"])
  print("目标存储桶 '%s' 已被移除" % (config["dest_bucket"]))

$ pip install opentelemetry-api
$ pip install opentelemetry-sdk

从 opentelemetry 导入 trace
从 opentelemetry.sdk.trace 导入 TracerProvider
从 opentelemetry.sdk.trace.export 导入 ConsoleSpanExporter
从 opentelemetry.sdk.trace.export 导入 BatchSpanProcessor
从 opentelemetry.sdk.resources 导入 SERVICE_NAME, Resource

resource = Resource(attributes={
  SERVICE_NAME: "my-minio"
})

provider = TracerProvider(resource=resource)
processor = BatchSpanProcessor(ConsoleSpanExporter())
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

tracer = trace.get_tracer(__name__)

使用 tracer.start_as_current_span("检查存储桶是否存在"):

from minio import Minio
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import ConsoleSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import SERVICE_NAME, Resource

# Convenient dict for basic config
config = {
  "dest_bucket":    "processed", # This will be auto created
  "minio_endpoint": "localhost:9000",
  "minio_username": "minio",
  "minio_password": "minioadmin",
}

# Initialize MinIO client
minio_client = Minio(config["minio_endpoint"],
              secure=False,
              access_key=config["minio_username"],
              secret_key=config["minio_password"]
              )

# 初始化 OpenTelemetry 提供程序
resource = Resource(attributes={
  SERVICE_NAME: "my-minio"
})

provider = TracerProvider(resource=resource)
processor = BatchSpanProcessor(ConsoleSpanExporter())
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

tracer = trace.get_tracer(__name__)

with tracer.start_as_current_span("检查存储桶是否存在"):
# 如果目标存储桶不存在，则创建
  if not minio_client.bucket_exists(config["dest_bucket"]):
    minio_client.make_bucket(config["dest_bucket"])
    print("目标存储桶 '%s' 已创建" % (config["dest_bucket"]))

...TRUNCATED...

$ python3 minio_app.py
Destination Bucket 'processed' has been created
<minio.datatypes.Object object at 0x103f36eb0>
Some objects here
Destination Bucket 'processed' has been removed
{
    "name": "check if bucket exists",
    "context": {
        "trace_id": "0xef41e07cf082045a2fc4eea70fd1a6de",
        "span_id": "0x867c14fe1fd97590",
        "trace_state": "[]"
    },
    "kind": "SpanKind.INTERNAL",
    "parent_id": null,
    "start_time": "2022-09-14T20:49:15.569511Z",
    "end_time": "2022-09-14T20:49:15.599886Z",
    "status": {
        "status_code": "UNSET"
    },
    "attributes": {},
    "events": [],
    "links": [],
    "resource": {
        "attributes": {
            "service.name": "my-minio"
        },
        "schema_url": ""
    }
}

使用 tracer.start_as_current_span(“检查存储桶是否存在”):
  current_span = trace.get_current_span()
  current_span.set_attribute(“function.name”, “CHECK_BUCKET”)

...（内容截断）…

   },
    “attributes”: {
        “function.name”: “CHECK_BUCKET”
    },
    “events”: [],

...TRUNCATED...

  current_span.add_event(“正在检查存储桶是否存在。”)
  如果 不 minio_client.bucket_exists(config[“dest_bucket”]):
    current_span.add_event(“存储桶不存在，将创建它。”)

    minio_client.make_bucket(config[“dest_bucket”])

...（内容截断）… 

"events":[
        {
            "name": "检查存储桶是否存在。",
            "timestamp": "2022-09-14T21:09:48.505709Z",
            "attributes": {}
        },
        {
            "name": "存储桶不存在，将创建它。",
            "timestamp": "2022-09-14T21:09:48.514541Z",
            "attributes": {}
        }
    ],

...TRUNCATED...

from minio import Minio
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import ConsoleSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import SERVICE_NAME, Resource

# Convenient dict for basic config
config = {
  "dest_bucket":    "processed", # This will be auto created
  "minio_endpoint": "localhost:9000",
  "minio_username": "minio",
  "minio_password": "minioadmin",
}

# Initialize MinIO client
minio_client = Minio(config["minio_endpoint"],
              secure=False,
              access_key=config["minio_username"],
              secret_key=config["minio_password"]
              )

resource = Resource(attributes={
  SERVICE_NAME: "my-minio"
})


provider = TracerProvider(resource=resource)
processor = BatchSpanProcessor(ConsoleSpanExporter())
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)


tracer = trace.get_tracer(__name__)


# Create destination bucket if it does not exist
with tracer.start_as_current_span("check if bucket exists"):
  current_span = trace.get_current_span()
  current_span.set_attribute("function.name", "CHECK_BUCKET")

  current_span.add_event("Checking if bucket exists.")
  if not minio_client.bucket_exists(config["dest_bucket"]):
    current_span.add_event("Bucket does not exist, going to create it.")

    with tracer.start_as_current_span("create bucket"):
      minio_client.make_bucket(config["dest_bucket"])
      current_span.add_event("Bucket has been created.")
      print("Destination Bucket '%s' has been created" % (config["dest_bucket"]))

with tracer.start_as_current_span("create object to add"):
  current_span = trace.get_current_span()
  current_span.set_attribute("function.name", "CREATE_OBJECT")

  # Create a test object
  file_path = "test_object.txt"
  f = open(file_path, "w")
  f.write("created test object")
  f.close()
  current_span.add_event("Test object has been created.")

# Put an object inside the bucket
with tracer.start_as_current_span("add created object to bucket"):
  current_span = trace.get_current_span()
  current_span.set_attribute("function.name", "CREATE_OBJECT")

  minio_client.fput_object(config["dest_bucket"], file_path, file_path)
  current_span.add_event("Test object has been placed in bucket.")

# Get the object from the bucket
with tracer.start_as_current_span("fetch object from bucket"):
  current_span = trace.get_current_span()
  current_span.set_attribute("function.name", "FETCH_OBJECT")

  minio_client.fget_object(config["dest_bucket"], file_path, config["dest_bucket"] + "/" + file_path)
  current_span.add_event("Test object has been fetched from bucket.")

# Get list of objects
for obj in minio_client.list_objects(config["dest_bucket"]):
  print(obj)
  print("Some objects here")


# Remove the object from bucket
with tracer.start_as_current_span("remove object from bucket"):
  current_span = trace.get_current_span()
  current_span.set_attribute("function.name", "REMOVE_OBJECT")

  minio_client.remove_object(config["dest_bucket"], file_path)
  current_span.add_event("Test object has been removed from bucket.")

# Get list of objects
for obj in minio_client.list_objects(config["dest_bucket"]):
  print(obj)
  print("No objects here")

# Remove destination bucket if it does exist
with tracer.start_as_current_span("check if bucket exists"):
  current_span = trace.get_current_span()
  current_span.set_attribute("function.name", "REMOVE_BUCKET")

  current_span.add_event("Checking if bucket exists.")
  if minio_client.bucket_exists(config["dest_bucket"]):
    current_span.add_event("Bucket exists, going to remove it.")

    with tracer.start_as_current_span("delete bucket"):
      minio_client.remove_bucket(config["dest_bucket"])
      current_span.add_event("Bucket has been removed.")
      print("Destination Bucket '%s' has been removed" % (config["dest_bucket"]))

$ docker run -d --name jaeger -p 16686:16686 -p 6831:6831/udp jaegertracing/all-in-one

$ pip install opentelemetry-exporter-jaeger

from opentelemetry.sdk.trace.export import ConsoleSpanExporter

from opentelemetry.exporter.jaeger.thrift import JaegerExporter

jaeger_exporter = JaegerExporter(
    agent_host_name="localhost",
    agent_port=6831,
)

processor = BatchSpanProcessor(ConsoleSpanExporter())

processor = BatchSpanProcessor(jaeger_exporter)

...（内容截断）…

from opentelemetry.exporter.jaeger.thrift import JaegerExporter

...TRUNCATED...

jaeger_exporter = JaegerExporter(
    agent_host_name="localhost",
    agent_port=6831,
)

provider = TracerProvider(resource=resource)
processor = BatchSpanProcessor(jaeger_exporter)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

tracer = trace.get_tracer(__name__)

...TRUNCATED...