diff --git a/api/.env.example b/api/.env.example index 2adde29d334..3e600365806 100644 --- a/api/.env.example +++ b/api/.env.example @@ -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 diff --git a/api/configs/feature/__init__.py b/api/configs/feature/__init__.py index c28716e3b0e..70c629b8070 100644 --- a/api/configs/feature/__init__.py +++ b/api/configs/feature/__init__.py @@ -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, diff --git a/api/extensions/ext_celery.py b/api/extensions/ext_celery.py index 2cf3505e918..690fa64cdf8 100644 --- a/api/extensions/ext_celery.py +++ b/api/extensions/ext_celery.py @@ -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) diff --git a/api/migrations/versions/2026_07_23_1200-6f5a9c2d8e1b_add_telemetry_fields_to_dify_setups.py b/api/migrations/versions/2026_07_23_1200-6f5a9c2d8e1b_add_telemetry_fields_to_dify_setups.py new file mode 100644 index 00000000000..ca5ad4a1608 --- /dev/null +++ b/api/migrations/versions/2026_07_23_1200-6f5a9c2d8e1b_add_telemetry_fields_to_dify_setups.py @@ -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") diff --git a/api/models/model.py b/api/models/model.py index c9f27a78b91..bcefb1c22fd 100644 --- a/api/models/model.py +++ b/api/models/model.py @@ -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 ) diff --git a/api/services/account_service.py b/api/services/account_service.py index cc2d983c4ef..cd89ddba2ab 100644 --- a/api/services/account_service.py +++ b/api/services/account_service.py @@ -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, diff --git a/api/services/telemetry_service.py b/api/services/telemetry_service.py new file mode 100644 index 00000000000..e6d141fe55a --- /dev/null +++ b/api/services/telemetry_service.py @@ -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" diff --git a/api/tasks/community_telemetry_task.py b/api/tasks/community_telemetry_task.py new file mode 100644 index 00000000000..c0c6eb46a88 --- /dev/null +++ b/api/tasks/community_telemetry_task.py @@ -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) diff --git a/api/tests/unit_tests/extensions/test_community_telemetry_celery.py b/api/tests/unit_tests/extensions/test_community_telemetry_celery.py new file mode 100644 index 00000000000..58ec7641457 --- /dev/null +++ b/api/tests/unit_tests/extensions/test_community_telemetry_celery.py @@ -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) diff --git a/api/tests/unit_tests/services/test_account_service.py b/api/tests/unit_tests/services/test_account_service.py index ad40dab358c..e7288909a16 100644 --- a/api/tests/unit_tests/services/test_account_service.py +++ b/api/tests/unit_tests/services/test_account_service.py @@ -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.""" diff --git a/api/tests/unit_tests/services/test_telemetry_service.py b/api/tests/unit_tests/services/test_telemetry_service.py new file mode 100644 index 00000000000..bec9dae012e --- /dev/null +++ b/api/tests/unit_tests/services/test_telemetry_service.py @@ -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 diff --git a/api/tests/unit_tests/tasks/test_community_telemetry_task.py b/api/tests/unit_tests/tasks/test_community_telemetry_task.py new file mode 100644 index 00000000000..7367b420483 --- /dev/null +++ b/api/tests/unit_tests/tasks/test_community_telemetry_task.py @@ -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)