mirror of
https://github.com/langgenius/dify.git
synced 2026-07-27 15:08:35 +08:00
feat: add CE telemetry report (#39452)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
FFXN
GitHub
parent
d5c0e927c6
commit
858b3b74dc
@ -729,7 +729,6 @@ OTEL_MAX_EXPORT_BATCH_SIZE=512
|
||||
OTEL_METRIC_EXPORT_INTERVAL=60000
|
||||
OTEL_BATCH_EXPORT_TIMEOUT=10000
|
||||
OTEL_METRIC_EXPORT_TIMEOUT=30000
|
||||
|
||||
# Prevent Clickjacking
|
||||
ALLOW_EMBED=false
|
||||
|
||||
|
||||
@ -816,6 +816,41 @@ class UpdateConfig(BaseSettings):
|
||||
)
|
||||
|
||||
|
||||
class CommunityTelemetryConfig(BaseSettings):
|
||||
"""
|
||||
Configuration for anonymous self-hosted community telemetry.
|
||||
"""
|
||||
|
||||
DISABLE_TELEMETRY: bool = Field(
|
||||
description="Disable anonymous community telemetry",
|
||||
default=False,
|
||||
)
|
||||
DO_NOT_TRACK: bool = Field(
|
||||
description="Respect the standard do-not-track opt-out signal for telemetry",
|
||||
default=False,
|
||||
)
|
||||
TELEMETRY_ENDPOINT: str = Field(
|
||||
description="Endpoint for anonymous community telemetry events",
|
||||
default="https://otel.dify.ai/v1/events",
|
||||
)
|
||||
TELEMETRY_FALLBACK_ENDPOINT: str = Field(
|
||||
description="Fallback endpoint for anonymous community telemetry events",
|
||||
default="https://otel.dify.cn/v1/events",
|
||||
)
|
||||
TELEMETRY_TIMEOUT_SECONDS: PositiveInt = Field(
|
||||
description="HTTP timeout in seconds for anonymous community telemetry requests",
|
||||
default=3,
|
||||
)
|
||||
TELEMETRY_HEARTBEAT_INTERVAL_MINUTES: PositiveInt = Field(
|
||||
description="Celery beat interval in minutes for checking whether heartbeat telemetry is due",
|
||||
default=30,
|
||||
)
|
||||
CI: bool = Field(
|
||||
description="Whether the process is running in CI; telemetry is skipped when true",
|
||||
default=False,
|
||||
)
|
||||
|
||||
|
||||
class WorkflowVariableTruncationConfig(BaseSettings):
|
||||
WORKFLOW_VARIABLE_TRUNCATION_MAX_SIZE: PositiveInt = Field(
|
||||
# 1000 KiB
|
||||
@ -1599,6 +1634,7 @@ class FeatureConfig(
|
||||
TenantIsolatedTaskQueueConfig,
|
||||
ToolConfig,
|
||||
UpdateConfig,
|
||||
CommunityTelemetryConfig,
|
||||
WorkflowConfig,
|
||||
WorkflowNodeExecutionConfig,
|
||||
WorkspaceConfig,
|
||||
|
||||
@ -5,6 +5,7 @@ from typing import Any
|
||||
import pytz # type: ignore[import-untyped]
|
||||
from celery import Celery, Task
|
||||
from celery.schedules import crontab
|
||||
from celery.signals import beat_init
|
||||
from typing_extensions import TypedDict
|
||||
|
||||
from configs import dify_config
|
||||
@ -36,6 +37,19 @@ class CeleryBeatScheduleEntry(TypedDict):
|
||||
schedule: crontab | timedelta
|
||||
|
||||
|
||||
def _enqueue_initial_community_telemetry_heartbeat(sender: Any, **_: Any) -> None:
|
||||
task_name = "community_telemetry.send_heartbeat"
|
||||
if "community_telemetry_heartbeat" not in sender.app.conf.beat_schedule:
|
||||
return
|
||||
|
||||
task = sender.app.tasks.get(task_name)
|
||||
if task is not None:
|
||||
task.apply_async()
|
||||
|
||||
|
||||
beat_init.connect(_enqueue_initial_community_telemetry_heartbeat, weak=False)
|
||||
|
||||
|
||||
def get_celery_ssl_options() -> CelerySSLOptionsDict | None:
|
||||
"""Get SSL configuration for Celery broker/backend connections."""
|
||||
# Only apply SSL if we're using Redis as broker/backend
|
||||
@ -260,6 +274,19 @@ def init_app(app: DifyApp) -> Celery:
|
||||
"schedule": timedelta(minutes=dify_config.API_TOKEN_LAST_USED_UPDATE_INTERVAL),
|
||||
}
|
||||
|
||||
if (
|
||||
dify_config.EDITION == "SELF_HOSTED"
|
||||
and not dify_config.ENTERPRISE_ENABLED
|
||||
and not dify_config.DISABLE_TELEMETRY
|
||||
and not dify_config.DO_NOT_TRACK
|
||||
and not dify_config.CI
|
||||
):
|
||||
imports.append("tasks.community_telemetry_task")
|
||||
beat_schedule["community_telemetry_heartbeat"] = {
|
||||
"task": "community_telemetry.send_heartbeat",
|
||||
"schedule": timedelta(minutes=dify_config.TELEMETRY_HEARTBEAT_INTERVAL_MINUTES),
|
||||
}
|
||||
|
||||
if dify_config.ENTERPRISE_ENABLED and dify_config.ENTERPRISE_TELEMETRY_ENABLED:
|
||||
imports.append("tasks.enterprise_telemetry_task")
|
||||
celery_app.conf.update(beat_schedule=beat_schedule, imports=imports)
|
||||
|
||||
@ -0,0 +1,30 @@
|
||||
"""add telemetry fields to dify_setups
|
||||
|
||||
Revision ID: 6f5a9c2d8e1b
|
||||
Revises: d2825e7b9c10
|
||||
Create Date: 2026-07-23 12:00:00.000000
|
||||
|
||||
"""
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision = "6f5a9c2d8e1b"
|
||||
down_revision = "d2825e7b9c10"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade():
|
||||
with op.batch_alter_table("dify_setups", schema=None) as batch_op:
|
||||
batch_op.add_column(sa.Column("instance_id", sa.String(length=255), nullable=True))
|
||||
batch_op.add_column(sa.Column("install_reported_at", sa.DateTime(), nullable=True))
|
||||
batch_op.add_column(sa.Column("last_heartbeat_at", sa.DateTime(), nullable=True))
|
||||
|
||||
|
||||
def downgrade():
|
||||
with op.batch_alter_table("dify_setups", schema=None) as batch_op:
|
||||
batch_op.drop_column("last_heartbeat_at")
|
||||
batch_op.drop_column("install_reported_at")
|
||||
batch_op.drop_column("instance_id")
|
||||
@ -362,6 +362,9 @@ class DifySetup(TypeBase):
|
||||
__table_args__ = (sa.PrimaryKeyConstraint("version", name="dify_setup_pkey"),)
|
||||
|
||||
version: Mapped[str] = mapped_column(String(255), nullable=False)
|
||||
instance_id: Mapped[str | None] = mapped_column(String(255), nullable=True, default=None)
|
||||
install_reported_at: Mapped[datetime | None] = mapped_column(sa.DateTime, nullable=True, default=None)
|
||||
last_heartbeat_at: Mapped[datetime | None] = mapped_column(sa.DateTime, nullable=True, default=None)
|
||||
setup_at: Mapped[datetime] = mapped_column(
|
||||
sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
|
||||
)
|
||||
|
||||
@ -75,6 +75,7 @@ from services.errors.account import (
|
||||
from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkspacesLimitExceededError
|
||||
from services.feature_service import FeatureService
|
||||
from services.plugin.plugin_auto_upgrade_service import PluginAutoUpgradeService
|
||||
from services.telemetry_service import CommunityTelemetryService
|
||||
from tasks.delete_account_task import delete_account_task
|
||||
from tasks.mail_account_deletion_task import send_account_deletion_verification_code
|
||||
from tasks.mail_change_mail_task import (
|
||||
@ -1953,7 +1954,7 @@ class RegisterService:
|
||||
|
||||
TenantService.create_owner_tenant_if_not_exist(account=account, is_setup=True, session=session)
|
||||
|
||||
dify_setup = DifySetup(version=dify_config.project.version)
|
||||
dify_setup = DifySetup(version=dify_config.project.version, instance_id=str(uuid.uuid4()))
|
||||
session.add(dify_setup)
|
||||
session.commit()
|
||||
except Exception as e:
|
||||
@ -1966,6 +1967,11 @@ class RegisterService:
|
||||
logger.exception("Setup account failed, email: %s, name: %s", email, name)
|
||||
raise ValueError(f"Setup failed: {e}")
|
||||
|
||||
try:
|
||||
CommunityTelemetryService.report_install(session=session)
|
||||
except Exception:
|
||||
logger.debug("Failed to report install telemetry", exc_info=True)
|
||||
|
||||
@classmethod
|
||||
def register(
|
||||
cls,
|
||||
|
||||
165
api/services/telemetry_service.py
Normal file
165
api/services/telemetry_service.py
Normal file
@ -0,0 +1,165 @@
|
||||
import logging
|
||||
import platform
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from typing import Literal
|
||||
|
||||
import httpx
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from configs import dify_config
|
||||
from libs.datetime_utils import naive_utc_now
|
||||
from models.model import DifySetup
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
TelemetryEvent = Literal["install", "heartbeat"]
|
||||
|
||||
SCHEMA_VERSION = 1
|
||||
|
||||
|
||||
class CommunityTelemetryService:
|
||||
@classmethod
|
||||
def report_install(cls, *, session: Session) -> bool:
|
||||
setup = cls._get_setup(session)
|
||||
if setup is None:
|
||||
return False
|
||||
|
||||
if setup.instance_id is None:
|
||||
setup.instance_id = str(uuid.uuid4())
|
||||
session.add(setup)
|
||||
session.commit()
|
||||
|
||||
payload = cls._build_payload(setup, "install")
|
||||
if not cls._send_event(payload):
|
||||
return False
|
||||
|
||||
setup.install_reported_at = naive_utc_now()
|
||||
session.add(setup)
|
||||
session.commit()
|
||||
return True
|
||||
|
||||
@classmethod
|
||||
def report_heartbeat(cls, *, session: Session, now: datetime | None = None) -> bool:
|
||||
setup = cls._get_setup(session)
|
||||
if setup is None:
|
||||
return False
|
||||
|
||||
if setup.instance_id is None:
|
||||
setup.instance_id = str(uuid.uuid4())
|
||||
session.add(setup)
|
||||
session.commit()
|
||||
|
||||
now = now or naive_utc_now()
|
||||
if not cls._is_heartbeat_due(setup, now):
|
||||
return False
|
||||
|
||||
if setup.install_reported_at is None:
|
||||
cls.report_install(session=session)
|
||||
|
||||
payload = cls._build_payload(setup, "heartbeat")
|
||||
if not cls._send_event(payload):
|
||||
return False
|
||||
|
||||
setup.last_heartbeat_at = now
|
||||
session.add(setup)
|
||||
session.commit()
|
||||
return True
|
||||
|
||||
@classmethod
|
||||
def _get_setup(cls, session: Session) -> DifySetup | None:
|
||||
return session.scalar(select(DifySetup).order_by(DifySetup.setup_at.asc()).limit(1))
|
||||
|
||||
@classmethod
|
||||
def _is_enabled(cls) -> bool:
|
||||
return (
|
||||
dify_config.EDITION == "SELF_HOSTED"
|
||||
and not dify_config.ENTERPRISE_ENABLED
|
||||
and not dify_config.DISABLE_TELEMETRY
|
||||
and not dify_config.DO_NOT_TRACK
|
||||
and not dify_config.CI
|
||||
and bool(dify_config.TELEMETRY_ENDPOINT)
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _build_payload(cls, setup: DifySetup, event: TelemetryEvent) -> dict[str, str | int]:
|
||||
payload: dict[str, str | int] = {
|
||||
"event": event,
|
||||
"instance_id": setup.instance_id or "",
|
||||
"version": setup.version if event == "install" else dify_config.project.version,
|
||||
"edition": dify_config.EDITION,
|
||||
"deployment_type": "unknown",
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"os": cls._normalize_os(platform.system()),
|
||||
"arch": cls._normalize_arch(platform.machine()),
|
||||
"sent_at": cls._format_datetime(naive_utc_now()),
|
||||
}
|
||||
|
||||
if event == "install":
|
||||
payload["installed_at"] = cls._format_datetime(setup.setup_at)
|
||||
|
||||
return payload
|
||||
|
||||
@classmethod
|
||||
def _send_event(cls, payload: dict[str, str | int]) -> bool:
|
||||
if not cls._is_enabled():
|
||||
return False
|
||||
|
||||
endpoints = [dify_config.TELEMETRY_ENDPOINT]
|
||||
if dify_config.TELEMETRY_FALLBACK_ENDPOINT not in endpoints:
|
||||
endpoints.append(dify_config.TELEMETRY_FALLBACK_ENDPOINT)
|
||||
|
||||
for endpoint in endpoints:
|
||||
if not endpoint:
|
||||
continue
|
||||
|
||||
try:
|
||||
response = httpx.post(
|
||||
endpoint,
|
||||
json=payload,
|
||||
timeout=dify_config.TELEMETRY_TIMEOUT_SECONDS,
|
||||
)
|
||||
response.raise_for_status()
|
||||
return True
|
||||
except httpx.RequestError:
|
||||
logger.debug("Failed to send community telemetry event to %s", endpoint, exc_info=True)
|
||||
except httpx.HTTPStatusError:
|
||||
logger.debug("Community telemetry endpoint returned an error: %s", endpoint, exc_info=True)
|
||||
return False
|
||||
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
def _is_heartbeat_due(cls, setup: DifySetup, now: datetime) -> bool:
|
||||
if setup.instance_id is None:
|
||||
return False
|
||||
|
||||
if setup.last_heartbeat_at is not None and setup.last_heartbeat_at.date() >= now.date():
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def _format_datetime(value: datetime) -> str:
|
||||
return value.replace(microsecond=0).isoformat() + "Z"
|
||||
|
||||
@staticmethod
|
||||
def _normalize_os(value: str) -> str:
|
||||
os_name = value.lower()
|
||||
if os_name in {"linux", "darwin", "windows"}:
|
||||
return os_name
|
||||
return "unknown"
|
||||
|
||||
@staticmethod
|
||||
def _normalize_arch(value: str) -> str:
|
||||
arch = value.lower()
|
||||
if arch in {"x86_64", "amd64"}:
|
||||
return "amd64"
|
||||
if arch in {"aarch64", "arm64"}:
|
||||
return "arm64"
|
||||
if arch.startswith("arm"):
|
||||
return "arm"
|
||||
if arch in {"i386", "i686", "x86"}:
|
||||
return "386"
|
||||
return "unknown"
|
||||
19
api/tasks/community_telemetry_task.py
Normal file
19
api/tasks/community_telemetry_task.py
Normal file
@ -0,0 +1,19 @@
|
||||
import logging
|
||||
|
||||
from celery import shared_task
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from extensions.ext_database import db
|
||||
from services.telemetry_service import CommunityTelemetryService
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@shared_task(name="community_telemetry.send_heartbeat", queue="schedule_executor")
|
||||
def send_community_telemetry_heartbeat() -> None:
|
||||
session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
|
||||
with session_factory() as session:
|
||||
try:
|
||||
CommunityTelemetryService.report_heartbeat(session=session)
|
||||
except Exception:
|
||||
logger.debug("Failed to process community telemetry heartbeat", exc_info=True)
|
||||
@ -0,0 +1,43 @@
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import Mock
|
||||
|
||||
from extensions.ext_celery import _enqueue_initial_community_telemetry_heartbeat
|
||||
|
||||
|
||||
def test_beat_start_enqueues_community_telemetry_heartbeat() -> None:
|
||||
task = Mock()
|
||||
sender = SimpleNamespace(
|
||||
app=SimpleNamespace(
|
||||
conf=SimpleNamespace(beat_schedule={"community_telemetry_heartbeat": {}}),
|
||||
tasks={"community_telemetry.send_heartbeat": task},
|
||||
)
|
||||
)
|
||||
|
||||
_enqueue_initial_community_telemetry_heartbeat(sender)
|
||||
|
||||
task.apply_async.assert_called_once_with()
|
||||
|
||||
|
||||
def test_beat_start_skips_community_telemetry_when_not_scheduled() -> None:
|
||||
task = Mock()
|
||||
sender = SimpleNamespace(
|
||||
app=SimpleNamespace(
|
||||
conf=SimpleNamespace(beat_schedule={}),
|
||||
tasks={"community_telemetry.send_heartbeat": task},
|
||||
)
|
||||
)
|
||||
|
||||
_enqueue_initial_community_telemetry_heartbeat(sender)
|
||||
|
||||
task.apply_async.assert_not_called()
|
||||
|
||||
|
||||
def test_beat_start_skips_community_telemetry_when_task_is_unavailable() -> None:
|
||||
sender = SimpleNamespace(
|
||||
app=SimpleNamespace(
|
||||
conf=SimpleNamespace(beat_schedule={"community_telemetry_heartbeat": {}}),
|
||||
tasks={},
|
||||
)
|
||||
)
|
||||
|
||||
_enqueue_initial_community_telemetry_heartbeat(sender)
|
||||
@ -2,6 +2,7 @@ import json
|
||||
from collections.abc import Iterator
|
||||
from datetime import datetime, timedelta
|
||||
from unittest.mock import MagicMock, patch
|
||||
from uuid import UUID
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import event, select
|
||||
@ -1325,7 +1326,10 @@ class TestRegisterService:
|
||||
with patch("services.account_service.AccountService.create_account") as mock_create_account:
|
||||
mock_create_account.return_value = mock_account
|
||||
|
||||
with patch("services.account_service.TenantService.create_owner_tenant_if_not_exist") as mock_create_tenant:
|
||||
with (
|
||||
patch("services.account_service.TenantService.create_owner_tenant_if_not_exist") as mock_create_tenant,
|
||||
patch("services.account_service.CommunityTelemetryService.report_install") as mock_report_install,
|
||||
):
|
||||
RegisterService.setup(
|
||||
"admin@example.com",
|
||||
"Admin User",
|
||||
@ -1344,7 +1348,39 @@ class TestRegisterService:
|
||||
session=sqlite_session,
|
||||
)
|
||||
mock_create_tenant.assert_called_once_with(account=mock_account, is_setup=True, session=sqlite_session)
|
||||
assert sqlite_session.scalar(select(DifySetup)) is not None
|
||||
dify_setup = sqlite_session.scalar(select(DifySetup))
|
||||
assert dify_setup is not None
|
||||
assert dify_setup.instance_id is not None
|
||||
assert str(UUID(dify_setup.instance_id)) == dify_setup.instance_id
|
||||
assert dify_setup.install_reported_at is None
|
||||
assert dify_setup.last_heartbeat_at is None
|
||||
mock_report_install.assert_called_once_with(session=sqlite_session)
|
||||
|
||||
def test_setup_succeeds_when_telemetry_install_report_fails(
|
||||
self, sqlite_session: Session, mock_external_service_dependencies
|
||||
):
|
||||
mock_external_service_dependencies["feature_service"].get_system_features.return_value.is_allow_register = True
|
||||
mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
|
||||
mock_account = TestAccountAssociatedDataFactory.create_account_mock()
|
||||
|
||||
with (
|
||||
patch("services.account_service.AccountService.create_account", return_value=mock_account),
|
||||
patch("services.account_service.TenantService.create_owner_tenant_if_not_exist"),
|
||||
patch(
|
||||
"services.account_service.CommunityTelemetryService.report_install",
|
||||
side_effect=RuntimeError("telemetry unavailable"),
|
||||
),
|
||||
):
|
||||
RegisterService.setup(
|
||||
"admin@example.com",
|
||||
"Admin User",
|
||||
"password123",
|
||||
"192.168.1.1",
|
||||
"en-US",
|
||||
session=sqlite_session,
|
||||
)
|
||||
|
||||
assert sqlite_session.scalar(select(DifySetup)) is not None
|
||||
|
||||
def test_setup_failure_rollback(self, sqlite_session: Session, mock_external_service_dependencies):
|
||||
"""Test setup failure with proper rollback."""
|
||||
|
||||
333
api/tests/unit_tests/services/test_telemetry_service.py
Normal file
333
api/tests/unit_tests/services/test_telemetry_service.py
Normal file
@ -0,0 +1,333 @@
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from unittest.mock import Mock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from models.model import DifySetup
|
||||
from services import telemetry_service
|
||||
from services.telemetry_service import CommunityTelemetryService
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def telemetry_enabled(monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.setattr(telemetry_service.dify_config, "EDITION", "SELF_HOSTED")
|
||||
monkeypatch.setattr(telemetry_service.dify_config, "ENTERPRISE_ENABLED", False)
|
||||
monkeypatch.setattr(telemetry_service.dify_config, "DISABLE_TELEMETRY", False)
|
||||
monkeypatch.setattr(telemetry_service.dify_config, "DO_NOT_TRACK", False)
|
||||
monkeypatch.setattr(telemetry_service.dify_config, "CI", False)
|
||||
monkeypatch.setattr(telemetry_service.dify_config, "TELEMETRY_ENDPOINT", "https://telemetry.example.test/v1/events")
|
||||
monkeypatch.setattr(
|
||||
telemetry_service.dify_config,
|
||||
"TELEMETRY_FALLBACK_ENDPOINT",
|
||||
"https://telemetry-cn.example.test/v1/events",
|
||||
)
|
||||
monkeypatch.setattr(telemetry_service.dify_config, "TELEMETRY_TIMEOUT_SECONDS", 2)
|
||||
|
||||
|
||||
def test_telemetry_is_disabled_for_enterprise(monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.setattr(telemetry_service.dify_config, "EDITION", "SELF_HOSTED")
|
||||
monkeypatch.setattr(telemetry_service.dify_config, "ENTERPRISE_ENABLED", True)
|
||||
|
||||
assert CommunityTelemetryService._is_enabled() is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("setting", "value"),
|
||||
[
|
||||
("EDITION", "CLOUD"),
|
||||
("DISABLE_TELEMETRY", True),
|
||||
("DO_NOT_TRACK", True),
|
||||
("CI", True),
|
||||
("TELEMETRY_ENDPOINT", ""),
|
||||
],
|
||||
)
|
||||
def test_telemetry_is_disabled_when_a_required_condition_is_not_met(
|
||||
telemetry_enabled, monkeypatch: pytest.MonkeyPatch, setting: str, value: str | bool
|
||||
):
|
||||
monkeypatch.setattr(telemetry_service.dify_config, setting, value)
|
||||
|
||||
assert CommunityTelemetryService._is_enabled() is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
|
||||
def test_reporting_without_setup_is_skipped(sqlite_session: Session, telemetry_enabled):
|
||||
assert CommunityTelemetryService.report_install(session=sqlite_session) is False
|
||||
assert CommunityTelemetryService.report_heartbeat(session=sqlite_session) is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
|
||||
def test_report_install_marks_reported_at(sqlite_session: Session, telemetry_enabled, monkeypatch: pytest.MonkeyPatch):
|
||||
setup = DifySetup(version="installed-version", instance_id="d246c3a1-350b-406c-92c7-6043df680758")
|
||||
sqlite_session.add(setup)
|
||||
sqlite_session.commit()
|
||||
monkeypatch.setattr(telemetry_service.dify_config.project, "version", "running-version")
|
||||
|
||||
sent_payloads: list[dict[str, str | int]] = []
|
||||
|
||||
def fake_post(url: str, json: dict[str, str | int], timeout: int):
|
||||
sent_payloads.append(json)
|
||||
return httpx.Response(204, request=httpx.Request("POST", url))
|
||||
|
||||
monkeypatch.setattr(telemetry_service.httpx, "post", fake_post)
|
||||
|
||||
assert CommunityTelemetryService.report_install(session=sqlite_session) is True
|
||||
|
||||
saved_setup = sqlite_session.scalar(select(DifySetup))
|
||||
assert saved_setup is not None
|
||||
assert saved_setup.install_reported_at is not None
|
||||
assert sent_payloads[0]["event"] == "install"
|
||||
assert sent_payloads[0]["instance_id"] == setup.instance_id
|
||||
assert sent_payloads[0]["version"] == "installed-version"
|
||||
assert "installed_at" in sent_payloads[0]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
|
||||
def test_report_install_generates_missing_instance_id(
|
||||
sqlite_session: Session, telemetry_enabled, monkeypatch: pytest.MonkeyPatch
|
||||
):
|
||||
setup = DifySetup(version="installed-version")
|
||||
sqlite_session.add(setup)
|
||||
sqlite_session.commit()
|
||||
monkeypatch.setattr(
|
||||
telemetry_service.httpx,
|
||||
"post",
|
||||
lambda url, json, timeout: httpx.Response(204, request=httpx.Request("POST", url)),
|
||||
)
|
||||
|
||||
assert CommunityTelemetryService.report_install(session=sqlite_session) is True
|
||||
|
||||
assert setup.instance_id is not None
|
||||
assert str(uuid.UUID(setup.instance_id)) == setup.instance_id
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
|
||||
def test_report_heartbeat_generates_missing_instance_id(
|
||||
sqlite_session: Session, telemetry_enabled, monkeypatch: pytest.MonkeyPatch
|
||||
):
|
||||
setup = DifySetup(version="1.0.0", install_reported_at=datetime(2026, 7, 12, 8, 0, 0))
|
||||
sqlite_session.add(setup)
|
||||
sqlite_session.commit()
|
||||
monkeypatch.setattr(
|
||||
telemetry_service.httpx,
|
||||
"post",
|
||||
lambda url, json, timeout: httpx.Response(204, request=httpx.Request("POST", url)),
|
||||
)
|
||||
|
||||
assert (
|
||||
CommunityTelemetryService.report_heartbeat(session=sqlite_session, now=datetime(2026, 7, 13, 12, 0, 0)) is True
|
||||
)
|
||||
|
||||
assert setup.instance_id is not None
|
||||
assert str(uuid.UUID(setup.instance_id)) == setup.instance_id
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
|
||||
def test_report_install_failure_keeps_install_pending(
|
||||
sqlite_session: Session, telemetry_enabled, monkeypatch: pytest.MonkeyPatch
|
||||
):
|
||||
setup = DifySetup(version="1.0.0", instance_id="d246c3a1-350b-406c-92c7-6043df680758")
|
||||
sqlite_session.add(setup)
|
||||
sqlite_session.commit()
|
||||
|
||||
def fake_post(url: str, json: dict[str, str | int], timeout: int):
|
||||
raise httpx.ConnectError("offline", request=httpx.Request("POST", url))
|
||||
|
||||
monkeypatch.setattr(telemetry_service.httpx, "post", fake_post)
|
||||
|
||||
assert CommunityTelemetryService.report_install(session=sqlite_session) is False
|
||||
|
||||
saved_setup = sqlite_session.scalar(select(DifySetup))
|
||||
assert saved_setup is not None
|
||||
assert saved_setup.install_reported_at is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
|
||||
def test_report_install_uses_fallback_endpoint_after_network_failure(
|
||||
sqlite_session: Session, telemetry_enabled, monkeypatch: pytest.MonkeyPatch
|
||||
):
|
||||
setup = DifySetup(version="1.0.0", instance_id="d246c3a1-350b-406c-92c7-6043df680758")
|
||||
sqlite_session.add(setup)
|
||||
sqlite_session.commit()
|
||||
|
||||
urls: list[str] = []
|
||||
|
||||
def fake_post(url: str, json: dict[str, str | int], timeout: int):
|
||||
urls.append(url)
|
||||
if url == telemetry_service.dify_config.TELEMETRY_ENDPOINT:
|
||||
raise httpx.ConnectError("offline", request=httpx.Request("POST", url))
|
||||
return httpx.Response(204, request=httpx.Request("POST", url))
|
||||
|
||||
monkeypatch.setattr(telemetry_service.httpx, "post", fake_post)
|
||||
|
||||
assert CommunityTelemetryService.report_install(session=sqlite_session) is True
|
||||
assert urls == [
|
||||
telemetry_service.dify_config.TELEMETRY_ENDPOINT,
|
||||
telemetry_service.dify_config.TELEMETRY_FALLBACK_ENDPOINT,
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
|
||||
def test_report_install_does_not_use_fallback_endpoint_after_http_error(
|
||||
sqlite_session: Session, telemetry_enabled, monkeypatch: pytest.MonkeyPatch
|
||||
):
|
||||
setup = DifySetup(version="1.0.0", instance_id="d246c3a1-350b-406c-92c7-6043df680758")
|
||||
sqlite_session.add(setup)
|
||||
sqlite_session.commit()
|
||||
|
||||
post_mock = Mock(
|
||||
return_value=httpx.Response(
|
||||
500,
|
||||
request=httpx.Request("POST", telemetry_service.dify_config.TELEMETRY_ENDPOINT),
|
||||
)
|
||||
)
|
||||
monkeypatch.setattr(telemetry_service.httpx, "post", post_mock)
|
||||
|
||||
assert CommunityTelemetryService.report_install(session=sqlite_session) is False
|
||||
post_mock.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
|
||||
def test_report_heartbeat_retries_pending_install_before_heartbeat(
|
||||
sqlite_session: Session, telemetry_enabled, monkeypatch: pytest.MonkeyPatch
|
||||
):
|
||||
setup = DifySetup(version="installed-version", instance_id="d246c3a1-350b-406c-92c7-6043df680758")
|
||||
sqlite_session.add(setup)
|
||||
sqlite_session.commit()
|
||||
monkeypatch.setattr(telemetry_service.dify_config.project, "version", "running-version")
|
||||
|
||||
sent_payloads: list[dict[str, str | int]] = []
|
||||
|
||||
def fake_post(url: str, json: dict[str, str | int], timeout: int):
|
||||
sent_payloads.append(json)
|
||||
return httpx.Response(204, request=httpx.Request("POST", url))
|
||||
|
||||
monkeypatch.setattr(telemetry_service.httpx, "post", fake_post)
|
||||
now = datetime(2026, 7, 13, 0, 0, 0)
|
||||
assert CommunityTelemetryService.report_heartbeat(session=sqlite_session, now=now) is True
|
||||
|
||||
saved_setup = sqlite_session.scalar(select(DifySetup))
|
||||
assert saved_setup is not None
|
||||
assert saved_setup.install_reported_at is not None
|
||||
assert saved_setup.last_heartbeat_at == now
|
||||
assert [(payload["event"], payload["version"]) for payload in sent_payloads] == [
|
||||
("install", "installed-version"),
|
||||
("heartbeat", "running-version"),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
|
||||
def test_report_heartbeat_skips_when_already_sent_today(
|
||||
sqlite_session: Session, telemetry_enabled, monkeypatch: pytest.MonkeyPatch
|
||||
):
|
||||
setup = DifySetup(
|
||||
version="1.0.0",
|
||||
instance_id="d246c3a1-350b-406c-92c7-6043df680758",
|
||||
install_reported_at=datetime(2026, 7, 13, 8, 0, 0),
|
||||
last_heartbeat_at=datetime(2026, 7, 13, 9, 0, 0),
|
||||
)
|
||||
sqlite_session.add(setup)
|
||||
sqlite_session.commit()
|
||||
|
||||
post_mock = Mock()
|
||||
monkeypatch.setattr(telemetry_service.httpx, "post", post_mock)
|
||||
|
||||
assert (
|
||||
CommunityTelemetryService.report_heartbeat(session=sqlite_session, now=datetime(2026, 7, 13, 12, 0, 0)) is False
|
||||
)
|
||||
post_mock.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
|
||||
def test_report_heartbeat_failure_does_not_mark_the_day_reported(
|
||||
sqlite_session: Session, telemetry_enabled, monkeypatch: pytest.MonkeyPatch
|
||||
):
|
||||
setup = DifySetup(
|
||||
version="1.0.0",
|
||||
instance_id="d246c3a1-350b-406c-92c7-6043df680758",
|
||||
install_reported_at=datetime(2026, 7, 13, 8, 0, 0),
|
||||
)
|
||||
sqlite_session.add(setup)
|
||||
sqlite_session.commit()
|
||||
|
||||
def fake_post(url: str, json: dict[str, str | int], timeout: int):
|
||||
raise httpx.ConnectError("offline", request=httpx.Request("POST", url))
|
||||
|
||||
monkeypatch.setattr(telemetry_service.httpx, "post", fake_post)
|
||||
|
||||
assert (
|
||||
CommunityTelemetryService.report_heartbeat(session=sqlite_session, now=datetime(2026, 7, 13, 12, 0, 0)) is False
|
||||
)
|
||||
assert setup.last_heartbeat_at is None
|
||||
|
||||
|
||||
def test_send_event_skips_when_telemetry_is_disabled(monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.setattr(telemetry_service.dify_config, "DISABLE_TELEMETRY", True)
|
||||
post_mock = Mock()
|
||||
monkeypatch.setattr(telemetry_service.httpx, "post", post_mock)
|
||||
|
||||
assert CommunityTelemetryService._send_event({"event": "heartbeat"}) is False
|
||||
post_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_send_event_skips_an_empty_fallback_endpoint(telemetry_enabled, monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.setattr(telemetry_service.dify_config, "TELEMETRY_FALLBACK_ENDPOINT", "")
|
||||
|
||||
def fake_post(url: str, json: dict[str, str], timeout: int):
|
||||
raise httpx.ConnectError("offline", request=httpx.Request("POST", url))
|
||||
|
||||
monkeypatch.setattr(telemetry_service.httpx, "post", fake_post)
|
||||
|
||||
assert CommunityTelemetryService._send_event({"event": "heartbeat"}) is False
|
||||
|
||||
|
||||
def test_send_event_does_not_retry_the_same_endpoint(telemetry_enabled, monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.setattr(
|
||||
telemetry_service.dify_config,
|
||||
"TELEMETRY_FALLBACK_ENDPOINT",
|
||||
telemetry_service.dify_config.TELEMETRY_ENDPOINT,
|
||||
)
|
||||
post_mock = Mock(
|
||||
return_value=httpx.Response(
|
||||
204,
|
||||
request=httpx.Request("POST", telemetry_service.dify_config.TELEMETRY_ENDPOINT),
|
||||
)
|
||||
)
|
||||
monkeypatch.setattr(telemetry_service.httpx, "post", post_mock)
|
||||
|
||||
assert CommunityTelemetryService._send_event({"event": "heartbeat"}) is True
|
||||
post_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_heartbeat_is_not_due_without_instance_id():
|
||||
setup = DifySetup(version="1.0.0")
|
||||
|
||||
assert CommunityTelemetryService._is_heartbeat_due(setup, datetime(2026, 7, 13, 12, 0, 0)) is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("value", "expected"),
|
||||
[
|
||||
("Linux", "linux"),
|
||||
("Plan9", "unknown"),
|
||||
],
|
||||
)
|
||||
def test_normalize_os(value: str, expected: str):
|
||||
assert CommunityTelemetryService._normalize_os(value) == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("value", "expected"),
|
||||
[
|
||||
("x86_64", "amd64"),
|
||||
("aarch64", "arm64"),
|
||||
("armv7l", "arm"),
|
||||
("i686", "386"),
|
||||
("riscv64", "unknown"),
|
||||
],
|
||||
)
|
||||
def test_normalize_arch(value: str, expected: str):
|
||||
assert CommunityTelemetryService._normalize_arch(value) == expected
|
||||
40
api/tests/unit_tests/tasks/test_community_telemetry_task.py
Normal file
40
api/tests/unit_tests/tasks/test_community_telemetry_task.py
Normal file
@ -0,0 +1,40 @@
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, Mock
|
||||
|
||||
import pytest
|
||||
|
||||
from tasks import community_telemetry_task
|
||||
|
||||
|
||||
def _configure_task_session(monkeypatch: pytest.MonkeyPatch) -> Mock:
|
||||
session = Mock()
|
||||
session_factory = MagicMock()
|
||||
session_factory.return_value.__enter__.return_value = session
|
||||
monkeypatch.setattr(community_telemetry_task, "db", SimpleNamespace(engine=object()))
|
||||
monkeypatch.setattr(community_telemetry_task, "sessionmaker", Mock(return_value=session_factory))
|
||||
return session
|
||||
|
||||
|
||||
def test_send_community_telemetry_heartbeat_reports_with_a_database_session(monkeypatch: pytest.MonkeyPatch):
|
||||
session = _configure_task_session(monkeypatch)
|
||||
report_heartbeat = Mock()
|
||||
monkeypatch.setattr(community_telemetry_task.CommunityTelemetryService, "report_heartbeat", report_heartbeat)
|
||||
|
||||
community_telemetry_task.send_community_telemetry_heartbeat.run()
|
||||
|
||||
report_heartbeat.assert_called_once_with(session=session)
|
||||
|
||||
|
||||
def test_send_community_telemetry_heartbeat_swallows_report_errors(monkeypatch: pytest.MonkeyPatch):
|
||||
_configure_task_session(monkeypatch)
|
||||
monkeypatch.setattr(
|
||||
community_telemetry_task.CommunityTelemetryService,
|
||||
"report_heartbeat",
|
||||
Mock(side_effect=RuntimeError("telemetry unavailable")),
|
||||
)
|
||||
log_debug = Mock()
|
||||
monkeypatch.setattr(community_telemetry_task.logger, "debug", log_debug)
|
||||
|
||||
community_telemetry_task.send_community_telemetry_heartbeat.run()
|
||||
|
||||
log_debug.assert_called_once_with("Failed to process community telemetry heartbeat", exc_info=True)
|
||||
Loading…
Reference in New Issue
Block a user