Добавлен REST API для мобильного приложения (этап 3, без Telegram)

Токен-аутентификация Sanctum (login/logout по устройствам),
эндпоинты /api/v1 для зон/устройств/правил автоматизации с тем же
RBAC (owner/viewer), что и в веб-версии — контроллеры переиспользуют
существующие Policy и FormRequest. Устройства отдают живой статус
из Device Shadow (Redis) и историю телеметрии из ClickHouse, плюс
управление (turn-on/turn-off/set-level) через device-control-service.

.env.example

@@ -43,10 +43,12 @@ INGEST_MQTT_TOPIC=devices/+/telemetry
 INGEST_BATCH_MAX_SIZE=500
 INGEST_BATCH_FLUSH_INTERVAL=5s
 INGEST_BATCH_FLUSH_TIMEOUT=10s
+INGEST_METRICS_PORT=9101
 
 # --- rule-engine-service ---
 DEVICE_CONTROL_HOST=device-control-service
 RULE_ENGINE_CACHE_REFRESH_INTERVAL=15s
+RULE_ENGINE_METRICS_PORT=9102
 
 # --- Laravel app (stage 2) ---
 # Regenerate for anything beyond local dev: php artisan key:generate --show
@@ -55,3 +57,9 @@ APP_ENV=local
 APP_DEBUG=true
 APP_URL=http://localhost:8000
 LARAVEL_HTTP_PORT=8000
+
+# --- Prometheus + Grafana (stage 2.5) ---
+PROMETHEUS_PORT=9090
+GRAFANA_PORT=3000
+GRAFANA_ADMIN_USER=admin
+GRAFANA_ADMIN_PASSWORD=change_me

docker-compose.yml

@@ -8,6 +8,8 @@ volumes:
   redis-data:
   rabbitmq-data:
   mosquitto-data:
+  prometheus-data:
+  grafana-data:
 
 services:
   mosquitto:
@@ -92,6 +94,8 @@ services:
 
   ingest-service:
     build: ./services/ingest-service
+    ports:
+      - "${INGEST_METRICS_PORT}:9101"
     environment:
       MQTT_HOST: mosquitto
       MQTT_PORT: 1883
@@ -104,6 +108,7 @@ services:
       RABBITMQ_PORT: 5672
       RABBITMQ_USER: ${RABBITMQ_USER}
       RABBITMQ_PASSWORD: ${RABBITMQ_PASSWORD}
+      INGEST_METRICS_PORT: 9101
     depends_on:
       mosquitto:
         condition: service_started
@@ -152,6 +157,8 @@ services:
     build:
       context: .
       dockerfile: services/rule-engine-service/Dockerfile
+    ports:
+      - "${RULE_ENGINE_METRICS_PORT}:9102"
     environment:
       POSTGRES_HOST: postgres
       POSTGRES_PORT: 5432
@@ -165,6 +172,7 @@ services:
       RABBITMQ_USER: ${RABBITMQ_USER}
       RABBITMQ_PASSWORD: ${RABBITMQ_PASSWORD}
       RULE_ENGINE_CACHE_REFRESH_INTERVAL: ${RULE_ENGINE_CACHE_REFRESH_INTERVAL}
+      RULE_ENGINE_METRICS_PORT: 9102
     depends_on:
       postgres:
         condition: service_healthy
@@ -222,5 +230,33 @@ services:
       - home-automation
     restart: unless-stopped
 
+  prometheus:
+    image: prom/prometheus:v3.1.0
+    ports:
+      - "${PROMETHEUS_PORT}:9090"
+    volumes:
+      - ./monitoring/prometheus/prometheus.yml:/etc/prometheus/prometheus.yml:ro
+      - prometheus-data:/prometheus
+    networks:
+      - home-automation
+    restart: unless-stopped
+
+  grafana:
+    image: grafana/grafana:11.4.0
+    ports:
+      - "${GRAFANA_PORT}:3000"
+    environment:
+      GF_SECURITY_ADMIN_USER: ${GRAFANA_ADMIN_USER}
+      GF_SECURITY_ADMIN_PASSWORD: ${GRAFANA_ADMIN_PASSWORD}
+      GF_USERS_ALLOW_SIGN_UP: "false"
+    volumes:
+      - grafana-data:/var/lib/grafana
+      - ./monitoring/grafana/provisioning:/etc/grafana/provisioning:ro
+    depends_on:
+      - prometheus
+    networks:
+      - home-automation
+    restart: unless-stopped
+
   # Remaining app services (health-check-service, esp32-emulator) are added
   # here as they get implemented — see services/*/README.md for the plan.

laravel-app/app/Http/Controllers/Api/V1/AuthController.php

@@ -0,0 +1,52 @@
+<?php
+
+namespace App\Http\Controllers\Api\V1;
+
+use App\Http\Controllers\Controller;
+use App\Http\Resources\UserResource;
+use App\Models\User;
+use Illuminate\Http\Request;
+use Illuminate\Support\Facades\Hash;
+use Illuminate\Validation\ValidationException;
+
+class AuthController extends Controller
+{
+    /**
+     * Issue a Sanctum personal access token for a mobile client. Each
+     * client names its own token (device_name) so a user can see/revoke
+     * per-device sessions later without logging everyone out at once.
+     */
+    public function login(Request $request)
+    {
+        $credentials = $request->validate([
+            'email' => ['required', 'email'],
+            'password' => ['required', 'string'],
+            'device_name' => ['required', 'string', 'max:255'],
+        ]);
+
+        $user = User::where('email', $credentials['email'])->first();
+
+        if (! $user || ! Hash::check($credentials['password'], $user->password)) {
+            throw ValidationException::withMessages([
+                'email' => ['Неверный email или пароль.'],
+            ]);
+        }
+
+        return response()->json([
+            'token' => $user->createToken($credentials['device_name'])->plainTextToken,
+            'user' => new UserResource($user),
+        ]);
+    }
+
+    public function logout(Request $request)
+    {
+        $request->user()->currentAccessToken()->delete();
+
+        return response()->noContent();
+    }
+
+    public function me(Request $request)
+    {
+        return new UserResource($request->user());
+    }
+}

laravel-app/app/Http/Controllers/Api/V1/AutomationRuleController.php

@@ -0,0 +1,66 @@
+<?php
+
+namespace App\Http\Controllers\Api\V1;
+
+use App\Http\Controllers\Controller;
+use App\Http\Requests\AutomationRuleRequest;
+use App\Http\Resources\AutomationRuleResource;
+use App\Models\AutomationRule;
+
+class AutomationRuleController extends Controller
+{
+    public function index()
+    {
+        $this->authorize('viewAny', AutomationRule::class);
+
+        $rules = AutomationRule::with(['zone', 'targetDevice', 'conditionSourceDevice'])
+            ->orderByDesc('id')
+            ->get();
+
+        return AutomationRuleResource::collection($rules);
+    }
+
+    public function store(AutomationRuleRequest $request)
+    {
+        $rule = new AutomationRule($this->mapActionParams($request->validated()));
+        $rule->user()->associate($request->user());
+        $rule->save();
+
+        return (new AutomationRuleResource($rule))->response()->setStatusCode(201);
+    }
+
+    public function update(AutomationRuleRequest $request, AutomationRule $automation_rule)
+    {
+        $automation_rule->update($this->mapActionParams($request->validated()));
+
+        return new AutomationRuleResource($automation_rule);
+    }
+
+    public function destroy(AutomationRule $automation_rule)
+    {
+        $this->authorize('delete', $automation_rule);
+
+        $automation_rule->delete();
+
+        return response()->noContent();
+    }
+
+    /**
+     * Mirrors AutomationRuleController@mapActionParams on the web side —
+     * "level" is a friendlier stand-in for action_params over the wire too,
+     * so mobile clients don't need to know the {"level": ...} shape.
+     *
+     * @param  array<string, mixed>  $data
+     * @return array<string, mixed>
+     */
+    private function mapActionParams(array $data): array
+    {
+        $data['action_params'] = $data['action_type'] === 'set_level'
+            ? ['level' => $data['level']]
+            : [];
+
+        unset($data['level']);
+
+        return $data;
+    }
+}

laravel-app/app/Http/Controllers/Api/V1/DeviceController.php

@@ -0,0 +1,126 @@
+<?php
+
+namespace App\Http\Controllers\Api\V1;
+
+use App\Http\Controllers\Controller;
+use App\Http\Requests\DeviceRequest;
+use App\Http\Resources\DeviceResource;
+use App\Models\Device;
+use App\Services\ClickHouseClient;
+use App\Services\DeviceControlClient;
+use App\Services\DeviceShadow;
+use Illuminate\Http\Request;
+
+class DeviceController extends Controller
+{
+    public function index(DeviceShadow $shadow)
+    {
+        $this->authorize('viewAny', Device::class);
+
+        $devices = Device::with(['zone', 'deviceType'])->orderBy('name')->get();
+        $statuses = $shadow->statuses($devices->pluck('external_id')->all());
+        $devices->each(fn (Device $d) => $d->live_status = $statuses[$d->external_id] ?? 'unknown');
+
+        return DeviceResource::collection($devices);
+    }
+
+    public function show(Device $device, DeviceShadow $shadow, ClickHouseClient $clickHouse)
+    {
+        $this->authorize('view', $device);
+
+        $device->load(['zone', 'deviceType']);
+        $snapshot = $shadow->snapshot($device->external_id);
+        $device->live_status = $snapshot['status'];
+
+        $telemetry = [];
+        if ($device->deviceType->category->value === 'sensor') {
+            $telemetry = $clickHouse->query(
+                'SELECT sensor_type, value, recorded_at FROM telemetry
+                 WHERE device_id = {device_id:String}
+                 ORDER BY recorded_at DESC LIMIT 50',
+                ['device_id' => $device->external_id],
+            );
+        }
+
+        return (new DeviceResource($device))->additional([
+            'shadow' => [
+                'last_seen' => $snapshot['last_seen'],
+                'desired_state' => $snapshot['desired_state'],
+                'reported_state' => $snapshot['reported_state'],
+            ],
+            'telemetry' => $telemetry,
+        ]);
+    }
+
+    public function store(DeviceRequest $request)
+    {
+        $device = new Device($request->validated());
+        $device->user()->associate($request->user());
+        $device->save();
+
+        return (new DeviceResource($device))->response()->setStatusCode(201);
+    }
+
+    public function update(DeviceRequest $request, Device $device)
+    {
+        $device->update($request->validated());
+
+        return new DeviceResource($device);
+    }
+
+    public function destroy(Device $device)
+    {
+        $this->authorize('delete', $device);
+
+        $device->delete();
+
+        return response()->noContent();
+    }
+
+    public function turnOn(Device $device, DeviceControlClient $client)
+    {
+        $this->authorize('update', $device);
+        $this->ensureCapability($device, 'turn_on');
+
+        return $this->respondToCommand($client->turnOn($device->external_id));
+    }
+
+    public function turnOff(Device $device, DeviceControlClient $client)
+    {
+        $this->authorize('update', $device);
+        $this->ensureCapability($device, 'turn_off');
+
+        return $this->respondToCommand($client->turnOff($device->external_id));
+    }
+
+    public function setLevel(Request $request, Device $device, DeviceControlClient $client)
+    {
+        $this->authorize('update', $device);
+        $this->ensureCapability($device, 'set_level');
+
+        $validated = $request->validate(['level' => ['required', 'numeric']]);
+
+        return $this->respondToCommand($client->setLevel($device->external_id, (float) $validated['level']));
+    }
+
+    private function ensureCapability(Device $device, string $capability): void
+    {
+        abort_unless(
+            in_array($capability, $device->deviceType->capabilities ?? [], true),
+            422,
+            "Устройство не поддерживает действие «{$capability}».",
+        );
+    }
+
+    /**
+     * @param  array{success: bool, error: ?string}  $result
+     */
+    private function respondToCommand(array $result)
+    {
+        if (! $result['success']) {
+            return response()->json(['message' => $result['error'] ?? 'Команда не выполнена.'], 422);
+        }
+
+        return response()->json(['message' => 'Команда отправлена.']);
+    }
+}

laravel-app/app/Http/Controllers/Api/V1/DeviceTypeController.php

@@ -0,0 +1,15 @@
+<?php
+
+namespace App\Http\Controllers\Api\V1;
+
+use App\Http\Controllers\Controller;
+use App\Http\Resources\DeviceTypeResource;
+use App\Models\DeviceType;
+
+class DeviceTypeController extends Controller
+{
+    public function index()
+    {
+        return DeviceTypeResource::collection(DeviceType::orderBy('code')->get());
+    }
+}

laravel-app/app/Http/Controllers/Api/V1/ZoneController.php

@@ -0,0 +1,41 @@
+<?php
+
+namespace App\Http\Controllers\Api\V1;
+
+use App\Http\Controllers\Controller;
+use App\Http\Requests\ZoneRequest;
+use App\Http\Resources\ZoneResource;
+use App\Models\Zone;
+
+class ZoneController extends Controller
+{
+    public function index()
+    {
+        $this->authorize('viewAny', Zone::class);
+
+        return ZoneResource::collection(Zone::withCount('devices')->orderBy('name')->get());
+    }
+
+    public function store(ZoneRequest $request)
+    {
+        $zone = $request->user()->zones()->create($request->validated());
+
+        return (new ZoneResource($zone))->response()->setStatusCode(201);
+    }
+
+    public function update(ZoneRequest $request, Zone $zone)
+    {
+        $zone->update($request->validated());
+
+        return new ZoneResource($zone);
+    }
+
+    public function destroy(Zone $zone)
+    {
+        $this->authorize('delete', $zone);
+
+        $zone->delete();
+
+        return response()->noContent();
+    }
+}

laravel-app/app/Http/Resources/AutomationRuleResource.php

@@ -0,0 +1,29 @@
+<?php
+
+namespace App\Http\Resources;
+
+use App\Models\AutomationRule;
+use Illuminate\Http\Request;
+use Illuminate\Http\Resources\Json\JsonResource;
+
+/** @mixin AutomationRule */
+class AutomationRuleResource extends JsonResource
+{
+    /** @return array<string, mixed> */
+    public function toArray(Request $request): array
+    {
+        return [
+            'id' => $this->id,
+            'zone' => new ZoneResource($this->whenLoaded('zone')),
+            'target_device' => new DeviceResource($this->whenLoaded('targetDevice')),
+            'condition_source_device' => new DeviceResource($this->whenLoaded('conditionSourceDevice')),
+            'condition_sensor_type' => $this->condition_sensor_type,
+            'condition_operator' => $this->condition_operator->value,
+            'condition_value' => $this->condition_value,
+            'action_type' => $this->action_type,
+            'action_params' => $this->action_params,
+            'is_active' => $this->is_active,
+            'created_at' => $this->created_at,
+        ];
+    }
+}

laravel-app/app/Http/Resources/DeviceResource.php

@@ -0,0 +1,33 @@
+<?php
+
+namespace App\Http\Resources;
+
+use App\Models\Device;
+use Illuminate\Http\Request;
+use Illuminate\Http\Resources\Json\JsonResource;
+
+/**
+ * @mixin Device
+ *
+ * `devices.status` in Postgres is only a stale snapshot (see migration
+ * comment) — Redis (Device Shadow) is authoritative. The controller stamps
+ * the live value onto the model as `live_status` before wrapping it here;
+ * `status` in the JSON always prefers that when present.
+ */
+class DeviceResource extends JsonResource
+{
+    /** @return array<string, mixed> */
+    public function toArray(Request $request): array
+    {
+        return [
+            'id' => $this->id,
+            'name' => $this->name,
+            'external_id' => $this->external_id,
+            'protocol' => $this->protocol,
+            'status' => $this->live_status ?? $this->status,
+            'zone' => new ZoneResource($this->whenLoaded('zone')),
+            'device_type' => new DeviceTypeResource($this->whenLoaded('deviceType')),
+            'created_at' => $this->created_at,
+        ];
+    }
+}

laravel-app/app/Http/Resources/DeviceTypeResource.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace App\Http\Resources;
+
+use App\Models\DeviceType;
+use Illuminate\Http\Request;
+use Illuminate\Http\Resources\Json\JsonResource;
+
+/** @mixin DeviceType */
+class DeviceTypeResource extends JsonResource
+{
+    /** @return array<string, mixed> */
+    public function toArray(Request $request): array
+    {
+        return [
+            'id' => $this->id,
+            'code' => $this->code,
+            'category' => $this->category->value,
+            'capabilities' => $this->capabilities,
+        ];
+    }
+}

laravel-app/app/Http/Resources/UserResource.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace App\Http\Resources;
+
+use App\Models\User;
+use Illuminate\Http\Request;
+use Illuminate\Http\Resources\Json\JsonResource;
+
+/** @mixin User */
+class UserResource extends JsonResource
+{
+    /** @return array<string, mixed> */
+    public function toArray(Request $request): array
+    {
+        return [
+            'id' => $this->id,
+            'name' => $this->name,
+            'email' => $this->email,
+            'role' => $this->role->value,
+        ];
+    }
+}

laravel-app/app/Http/Resources/ZoneResource.php

@@ -0,0 +1,23 @@
+<?php
+
+namespace App\Http\Resources;
+
+use App\Models\Zone;
+use Illuminate\Http\Request;
+use Illuminate\Http\Resources\Json\JsonResource;
+
+/** @mixin Zone */
+class ZoneResource extends JsonResource
+{
+    /** @return array<string, mixed> */
+    public function toArray(Request $request): array
+    {
+        return [
+            'id' => $this->id,
+            'name' => $this->name,
+            'description' => $this->description,
+            'devices_count' => $this->whenCounted('devices'),
+            'created_at' => $this->created_at,
+        ];
+    }
+}

laravel-app/routes/api.php

@@ -1,8 +1,38 @@
 <?php
 
-use Illuminate\Http\Request;
+use App\Http\Controllers\Api\V1\AuthController;
+use App\Http\Controllers\Api\V1\AutomationRuleController;
+use App\Http\Controllers\Api\V1\DeviceController;
+use App\Http\Controllers\Api\V1\DeviceTypeController;
+use App\Http\Controllers\Api\V1\ZoneController;
 use Illuminate\Support\Facades\Route;
 
-Route::get('/user', function (Request $request) {
+Route::prefix('v1')->group(function () {
-    return $request->user();
+    Route::post('/login', [AuthController::class, 'login']);
-})->middleware('auth:sanctum');
+
+    Route::middleware('auth:sanctum')->group(function () {
+        Route::post('/logout', [AuthController::class, 'logout']);
+        Route::get('/me', [AuthController::class, 'me']);
+
+        Route::get('/zones', [ZoneController::class, 'index']);
+        Route::post('/zones', [ZoneController::class, 'store']);
+        Route::put('/zones/{zone}', [ZoneController::class, 'update']);
+        Route::delete('/zones/{zone}', [ZoneController::class, 'destroy']);
+
+        Route::get('/device-types', [DeviceTypeController::class, 'index']);
+
+        Route::get('/devices', [DeviceController::class, 'index']);
+        Route::post('/devices', [DeviceController::class, 'store']);
+        Route::get('/devices/{device}', [DeviceController::class, 'show']);
+        Route::put('/devices/{device}', [DeviceController::class, 'update']);
+        Route::delete('/devices/{device}', [DeviceController::class, 'destroy']);
+        Route::post('/devices/{device}/turn-on', [DeviceController::class, 'turnOn']);
+        Route::post('/devices/{device}/turn-off', [DeviceController::class, 'turnOff']);
+        Route::post('/devices/{device}/set-level', [DeviceController::class, 'setLevel']);
+
+        Route::get('/automation-rules', [AutomationRuleController::class, 'index']);
+        Route::post('/automation-rules', [AutomationRuleController::class, 'store']);
+        Route::put('/automation-rules/{automation_rule}', [AutomationRuleController::class, 'update']);
+        Route::delete('/automation-rules/{automation_rule}', [AutomationRuleController::class, 'destroy']);
+    });
+});

laravel-app/tests/Feature/Api/V1/AuthTest.php

@@ -0,0 +1,82 @@
+<?php
+
+namespace Tests\Feature\Api\V1;
+
+use App\Enums\UserRole;
+use App\Models\User;
+use Illuminate\Foundation\Testing\RefreshDatabase;
+use Tests\TestCase;
+
+class AuthTest extends TestCase
+{
+    use RefreshDatabase;
+
+    public function test_login_issues_a_token(): void
+    {
+        $user = User::factory()->create([
+            'role' => UserRole::Owner,
+            'password' => bcrypt('secret1234'),
+        ]);
+
+        $response = $this->postJson('/api/v1/login', [
+            'email' => $user->email,
+            'password' => 'secret1234',
+            'device_name' => 'iphone-15',
+        ]);
+
+        $response->assertOk();
+        $response->assertJsonStructure(['token', 'user' => ['id', 'name', 'email', 'role']]);
+        $this->assertDatabaseHas('personal_access_tokens', [
+            'tokenable_id' => $user->id,
+            'name' => 'iphone-15',
+        ]);
+    }
+
+    public function test_login_rejects_wrong_password(): void
+    {
+        $user = User::factory()->create(['password' => bcrypt('secret1234')]);
+
+        $response = $this->postJson('/api/v1/login', [
+            'email' => $user->email,
+            'password' => 'wrong-password',
+            'device_name' => 'iphone-15',
+        ]);
+
+        $response->assertStatus(422);
+        $response->assertJsonValidationErrors('email');
+    }
+
+    public function test_authenticated_token_can_reach_protected_route(): void
+    {
+        $user = User::factory()->create(['password' => bcrypt('secret1234')]);
+
+        $token = $this->postJson('/api/v1/login', [
+            'email' => $user->email,
+            'password' => 'secret1234',
+            'device_name' => 'iphone-15',
+        ])->json('token');
+
+        $this->withHeader('Authorization', "Bearer {$token}")
+            ->getJson('/api/v1/me')
+            ->assertOk()
+            ->assertJsonPath('data.email', $user->email);
+    }
+
+    public function test_logout_revokes_the_current_token(): void
+    {
+        $user = User::factory()->create(['password' => bcrypt('secret1234')]);
+
+        $tokenModel = $user->createToken('iphone-15');
+
+        $this->withHeader('Authorization', "Bearer {$tokenModel->plainTextToken}")
+            ->postJson('/api/v1/logout')
+            ->assertNoContent();
+
+        $this->assertDatabaseMissing('personal_access_tokens', ['id' => $tokenModel->accessToken->id]);
+    }
+
+    public function test_guest_cannot_reach_protected_route(): void
+    {
+        $this->getJson('/api/v1/me')->assertUnauthorized();
+    }
+}

laravel-app/tests/Feature/Api/V1/AutomationRuleApiTest.php

@@ -0,0 +1,124 @@
+<?php
+
+namespace Tests\Feature\Api\V1;
+
+use App\Enums\DeviceCategory;
+use App\Enums\UserRole;
+use App\Models\Device;
+use App\Models\DeviceType;
+use App\Models\User;
+use App\Models\Zone;
+use Illuminate\Foundation\Testing\RefreshDatabase;
+use Tests\TestCase;
+
+class AutomationRuleApiTest extends TestCase
+{
+    use RefreshDatabase;
+
+    private function makeZoneWithDevices(User $owner): array
+    {
+        $zone = Zone::forceCreate(['user_id' => $owner->id, 'name' => 'Гроубокс']);
+
+        $sensorType = DeviceType::create([
+            'code' => 'sensor_temp_humidity',
+            'category' => DeviceCategory::Sensor,
+            'capabilities' => ['temperature', 'humidity'],
+        ]);
+        $fanType = DeviceType::create([
+            'code' => 'fan',
+            'category' => DeviceCategory::Actuator,
+            'capabilities' => ['turn_on', 'turn_off', 'set_level'],
+        ]);
+
+        $sensor = Device::forceCreate([
+            'user_id' => $owner->id, 'zone_id' => $zone->id, 'device_type_id' => $sensorType->id,
+            'name' => 'Датчик', 'external_id' => 'sensor-1', 'protocol' => 'mqtt',
+        ]);
+        $fan = Device::forceCreate([
+            'user_id' => $owner->id, 'zone_id' => $zone->id, 'device_type_id' => $fanType->id,
+            'name' => 'Вентилятор', 'external_id' => 'fan-1', 'protocol' => 'mqtt',
+        ]);
+
+        return [$zone, $sensor, $fan];
+    }
+
+    public function test_owner_can_create_rule_with_turn_on_action(): void
+    {
+        $owner = User::factory()->create(['role' => UserRole::Owner]);
+        [$zone, $sensor, $fan] = $this->makeZoneWithDevices($owner);
+
+        $response = $this->actingAs($owner)->postJson('/api/v1/automation-rules', [
+            'zone_id' => $zone->id,
+            'condition_source_device_id' => $sensor->id,
+            'condition_sensor_type' => 'temperature',
+            'condition_operator' => '>',
+            'condition_value' => 28,
+            'target_device_id' => $fan->id,
+            'action_type' => 'turn_on',
+            'is_active' => true,
+        ]);
+
+        $response->assertCreated();
+        // '{}', not '[]' — same PHP/Go empty-array-vs-object gotcha as the web controller.
+        $this->assertDatabaseHas('automation_rules', [
+            'target_device_id' => $fan->id,
+            'action_type' => 'turn_on',
+            'action_params' => '{}',
+        ]);
+    }
+
+    public function test_set_level_action_requires_and_stores_level_param(): void
+    {
+        $owner = User::factory()->create(['role' => UserRole::Owner]);
+        [$zone, $sensor, $fan] = $this->makeZoneWithDevices($owner);
+
+        $missingLevel = $this->actingAs($owner)->postJson('/api/v1/automation-rules', [
+            'zone_id' => $zone->id,
+            'condition_source_device_id' => $sensor->id,
+            'condition_sensor_type' => 'temperature',
+            'condition_operator' => '>',
+            'condition_value' => 28,
+            'target_device_id' => $fan->id,
+            'action_type' => 'set_level',
+        ]);
+        $missingLevel->assertJsonValidationErrors('level');
+
+        $response = $this->actingAs($owner)->postJson('/api/v1/automation-rules', [
+            'zone_id' => $zone->id,
+            'condition_source_device_id' => $sensor->id,
+            'condition_sensor_type' => 'temperature',
+            'condition_operator' => '>',
+            'condition_value' => 28,
+            'target_device_id' => $fan->id,
+            'action_type' => 'set_level',
+            'level' => 42,
+        ]);
+
+        $response->assertCreated();
+        $this->assertDatabaseHas('automation_rules', [
+            'target_device_id' => $fan->id,
+            'action_type' => 'set_level',
+            'action_params' => '{"level":42}',
+        ]);
+    }
+
+    public function test_viewer_cannot_create_rule(): void
+    {
+        $owner = User::factory()->create(['role' => UserRole::Owner]);
+        $viewer = User::factory()->create(['role' => UserRole::Viewer]);
+        [$zone, $sensor, $fan] = $this->makeZoneWithDevices($owner);
+
+        $response = $this->actingAs($viewer)->postJson('/api/v1/automation-rules', [
+            'zone_id' => $zone->id,
+            'condition_source_device_id' => $sensor->id,
+            'condition_sensor_type' => 'temperature',
+            'condition_operator' => '>',
+            'condition_value' => 28,
+            'target_device_id' => $fan->id,
+            'action_type' => 'turn_on',
+        ]);
+
+        $response->assertForbidden();
+        $this->assertDatabaseCount('automation_rules', 0);
+    }
+}

laravel-app/tests/Feature/Api/V1/DeviceApiTest.php

@@ -0,0 +1,138 @@
+<?php
+
+namespace Tests\Feature\Api\V1;
+
+use App\Enums\DeviceCategory;
+use App\Enums\UserRole;
+use App\Models\Device;
+use App\Models\DeviceType;
+use App\Models\User;
+use App\Models\Zone;
+use App\Services\ClickHouseClient;
+use App\Services\DeviceControlClient;
+use App\Services\DeviceShadow;
+use Illuminate\Foundation\Testing\RefreshDatabase;
+use Tests\TestCase;
+
+class DeviceApiTest extends TestCase
+{
+    use RefreshDatabase;
+
+    private function makeDevice(User $owner, array $capabilities, string $category = DeviceCategory::Actuator->value): Device
+    {
+        $zone = Zone::forceCreate(['user_id' => $owner->id, 'name' => 'Гроубокс']);
+        $type = DeviceType::create([
+            'code' => 'fan',
+            'category' => $category,
+            'capabilities' => $capabilities,
+        ]);
+
+        return Device::forceCreate([
+            'user_id' => $owner->id, 'zone_id' => $zone->id, 'device_type_id' => $type->id,
+            'name' => 'Вентилятор', 'external_id' => 'fan-1', 'protocol' => 'mqtt',
+        ]);
+    }
+
+    public function test_index_includes_live_status_from_shadow(): void
+    {
+        $owner = User::factory()->create(['role' => UserRole::Owner]);
+        $device = $this->makeDevice($owner, ['turn_on']);
+
+        $this->mock(DeviceShadow::class, function ($mock) {
+            $mock->shouldReceive('statuses')->once()->with(['fan-1'])->andReturn(['fan-1' => 'online']);
+        });
+
+        $response = $this->actingAs($owner)->getJson('/api/v1/devices');
+
+        $response->assertOk();
+        $response->assertJsonPath('data.0.status', 'online');
+        $response->assertJsonPath('data.0.external_id', $device->external_id);
+    }
+
+    public function test_show_includes_shadow_snapshot_and_telemetry(): void
+    {
+        $owner = User::factory()->create(['role' => UserRole::Owner]);
+        $device = $this->makeDevice($owner, ['temperature'], DeviceCategory::Sensor->value);
+
+        $this->mock(DeviceShadow::class, function ($mock) {
+            $mock->shouldReceive('snapshot')->once()->with('fan-1')->andReturn([
+                'status' => 'online',
+                'last_seen' => null,
+                'desired_state' => [],
+                'reported_state' => ['temperature' => 24.5],
+            ]);
+        });
+        $this->mock(ClickHouseClient::class, function ($mock) {
+            $mock->shouldReceive('query')->once()
+                ->andReturn([['sensor_type' => 'temperature', 'value' => 24.5, 'recorded_at' => '2026-01-01 00:00:00']]);
+        });
+
+        $response = $this->actingAs($owner)->getJson("/api/v1/devices/{$device->id}");
+
+        $response->assertOk();
+        $response->assertJsonPath('data.status', 'online');
+        $response->assertJsonPath('shadow.reported_state.temperature', 24.5);
+        $response->assertJsonPath('telemetry.0.sensor_type', 'temperature');
+    }
+
+    public function test_owner_can_turn_on_device(): void
+    {
+        $owner = User::factory()->create(['role' => UserRole::Owner]);
+        $device = $this->makeDevice($owner, ['turn_on', 'turn_off']);
+
+        $this->mock(DeviceControlClient::class, function ($mock) {
+            $mock->shouldReceive('turnOn')->once()->with('fan-1')
+                ->andReturn(['success' => true, 'error' => null]);
+        });
+
+        $response = $this->actingAs($owner)->postJson("/api/v1/devices/{$device->id}/turn-on");
+
+        $response->assertOk();
+        $response->assertJsonPath('message', 'Команда отправлена.');
+    }
+
+    public function test_turn_on_rejected_when_device_lacks_capability(): void
+    {
+        $owner = User::factory()->create(['role' => UserRole::Owner]);
+        $device = $this->makeDevice($owner, ['turn_off']); // no turn_on
+
+        $this->mock(DeviceControlClient::class, function ($mock) {
+            $mock->shouldNotReceive('turnOn');
+        });
+
+        $response = $this->actingAs($owner)->postJson("/api/v1/devices/{$device->id}/turn-on");
+
+        $response->assertStatus(422);
+    }
+
+    public function test_viewer_cannot_turn_on_device(): void
+    {
+        $owner = User::factory()->create(['role' => UserRole::Owner]);
+        $viewer = User::factory()->create(['role' => UserRole::Viewer]);
+        $device = $this->makeDevice($owner, ['turn_on']);
+
+        $this->mock(DeviceControlClient::class, function ($mock) {
+            $mock->shouldNotReceive('turnOn');
+        });
+
+        $response = $this->actingAs($viewer)->postJson("/api/v1/devices/{$device->id}/turn-on");
+
+        $response->assertForbidden();
+    }
+
+    public function test_failed_command_returns_422_with_message(): void
+    {
+        $owner = User::factory()->create(['role' => UserRole::Owner]);
+        $device = $this->makeDevice($owner, ['turn_on']);
+
+        $this->mock(DeviceControlClient::class, function ($mock) {
+            $mock->shouldReceive('turnOn')->once()
+                ->andReturn(['success' => false, 'error' => 'device offline']);
+        });
+
+        $response = $this->actingAs($owner)->postJson("/api/v1/devices/{$device->id}/turn-on");
+
+        $response->assertStatus(422);
+        $response->assertJsonPath('message', 'device offline');
+    }
+}

laravel-app/tests/Feature/Api/V1/ZoneApiTest.php

@@ -0,0 +1,65 @@
+<?php
+
+namespace Tests\Feature\Api\V1;
+
+use App\Enums\UserRole;
+use App\Models\User;
+use App\Models\Zone;
+use Illuminate\Foundation\Testing\RefreshDatabase;
+use Tests\TestCase;
+
+class ZoneApiTest extends TestCase
+{
+    use RefreshDatabase;
+
+    public function test_owner_can_create_zone(): void
+    {
+        $owner = User::factory()->create(['role' => UserRole::Owner]);
+
+        $response = $this->actingAs($owner)->postJson('/api/v1/zones', [
+            'name' => 'Гостиная',
+            'description' => 'Тест',
+        ]);
+
+        $response->assertCreated();
+        $response->assertJsonPath('data.name', 'Гостиная');
+        $this->assertDatabaseHas('zones', ['name' => 'Гостиная', 'user_id' => $owner->id]);
+    }
+
+    public function test_viewer_cannot_create_zone(): void
+    {
+        $viewer = User::factory()->create(['role' => UserRole::Viewer]);
+
+        $response = $this->actingAs($viewer)->postJson('/api/v1/zones', ['name' => 'Гостиная']);
+
+        $response->assertForbidden();
+        $this->assertDatabaseMissing('zones', ['name' => 'Гостиная']);
+    }
+
+    public function test_viewer_can_list_zones(): void
+    {
+        $viewer = User::factory()->create(['role' => UserRole::Viewer]);
+        Zone::forceCreate(['user_id' => $viewer->id, 'name' => 'Гроубокс']);
+
+        $response = $this->actingAs($viewer)->getJson('/api/v1/zones');
+
+        $response->assertOk();
+        $response->assertJsonPath('data.0.name', 'Гроубокс');
+    }
+
+    public function test_viewer_cannot_delete_zone(): void
+    {
+        $viewer = User::factory()->create(['role' => UserRole::Viewer]);
+        $zone = Zone::forceCreate(['user_id' => $viewer->id, 'name' => 'Гроубокс']);
+
+        $response = $this->actingAs($viewer)->deleteJson("/api/v1/zones/{$zone->id}");
+
+        $response->assertForbidden();
+        $this->assertDatabaseHas('zones', ['id' => $zone->id]);
+    }
+
+    public function test_guest_gets_401(): void
+    {
+        $this->getJson('/api/v1/zones')->assertUnauthorized();
+    }
+}

monitoring/grafana/provisioning/dashboards/automation-devices.json

@@ -0,0 +1,83 @@
+{
+  "uid": "automation-devices",
+  "title": "Automation & Devices",
+  "tags": ["home-automation"],
+  "timezone": "browser",
+  "schemaVersion": 39,
+  "version": 1,
+  "refresh": "10s",
+  "time": { "from": "now-30m", "to": "now" },
+  "panels": [
+    {
+      "id": 1,
+      "title": "Rules triggered (rate, by outcome)",
+      "type": "timeseries",
+      "datasource": { "type": "prometheus", "uid": "Prometheus" },
+      "gridPos": { "h": 8, "w": 12, "x": 0, "y": 0 },
+      "targets": [
+        {
+          "expr": "sum by (outcome) (rate(rule_engine_rules_triggered_total[1m]))",
+          "legendFormat": "{{outcome}}",
+          "refId": "A"
+        }
+      ]
+    },
+    {
+      "id": 2,
+      "title": "device-control-service dispatch duration (p95, by action)",
+      "type": "timeseries",
+      "datasource": { "type": "prometheus", "uid": "Prometheus" },
+      "gridPos": { "h": 8, "w": 12, "x": 12, "y": 0 },
+      "fieldConfig": { "defaults": { "unit": "s" }, "overrides": [] },
+      "targets": [
+        {
+          "expr": "histogram_quantile(0.95, sum(rate(rule_engine_dispatch_duration_seconds_bucket[5m])) by (le, action_type))",
+          "legendFormat": "{{action_type}}",
+          "refId": "A"
+        }
+      ]
+    },
+    {
+      "id": 3,
+      "title": "Device commands (rate, by action + outcome)",
+      "type": "timeseries",
+      "datasource": { "type": "prometheus", "uid": "Prometheus" },
+      "gridPos": { "h": 8, "w": 12, "x": 0, "y": 8 },
+      "targets": [
+        {
+          "expr": "sum by (action, outcome) (rate(device_control_commands_total[1m]))",
+          "legendFormat": "{{action}} / {{outcome}}",
+          "refId": "A"
+        }
+      ]
+    },
+    {
+      "id": 4,
+      "title": "Device online/offline transitions (rate)",
+      "type": "timeseries",
+      "datasource": { "type": "prometheus", "uid": "Prometheus" },
+      "gridPos": { "h": 8, "w": 12, "x": 12, "y": 8 },
+      "targets": [
+        {
+          "expr": "sum by (direction) (rate(device_control_health_transitions_total[5m]))",
+          "legendFormat": "{{direction}}",
+          "refId": "A"
+        }
+      ]
+    },
+    {
+      "id": 5,
+      "title": "Automation rule cache refreshes (rate, by outcome)",
+      "type": "timeseries",
+      "datasource": { "type": "prometheus", "uid": "Prometheus" },
+      "gridPos": { "h": 8, "w": 12, "x": 0, "y": 16 },
+      "targets": [
+        {
+          "expr": "sum by (outcome) (rate(rule_engine_cache_refresh_total[1m]))",
+          "legendFormat": "{{outcome}}",
+          "refId": "A"
+        }
+      ]
+    }
+  ]
+}

monitoring/grafana/provisioning/dashboards/dashboards.yml

@@ -0,0 +1,11 @@
+apiVersion: 1
+
+providers:
+  - name: default
+    orgId: 1
+    folder: ""
+    type: file
+    disableDeletion: false
+    updateIntervalSeconds: 30
+    options:
+      path: /etc/grafana/provisioning/dashboards

monitoring/grafana/provisioning/dashboards/ingest-telemetry.json

@@ -0,0 +1,88 @@
+{
+  "uid": "ingest-telemetry",
+  "title": "Ingest & Telemetry",
+  "tags": ["home-automation"],
+  "timezone": "browser",
+  "schemaVersion": 39,
+  "version": 1,
+  "refresh": "10s",
+  "time": { "from": "now-30m", "to": "now" },
+  "panels": [
+    {
+      "id": 1,
+      "title": "MQTT messages received (rate)",
+      "type": "timeseries",
+      "datasource": { "type": "prometheus", "uid": "Prometheus" },
+      "gridPos": { "h": 8, "w": 12, "x": 0, "y": 0 },
+      "targets": [
+        {
+          "expr": "sum(rate(ingest_mqtt_messages_received_total[1m]))",
+          "legendFormat": "received/s",
+          "refId": "A"
+        }
+      ]
+    },
+    {
+      "id": 2,
+      "title": "MQTT messages invalid / dropped (rate)",
+      "type": "timeseries",
+      "datasource": { "type": "prometheus", "uid": "Prometheus" },
+      "gridPos": { "h": 8, "w": 12, "x": 12, "y": 0 },
+      "targets": [
+        {
+          "expr": "sum(rate(ingest_mqtt_messages_invalid_total[1m]))",
+          "legendFormat": "invalid/s",
+          "refId": "A"
+        },
+        {
+          "expr": "sum(rate(ingest_mqtt_messages_dropped_total[1m]))",
+          "legendFormat": "dropped (backlog full)/s",
+          "refId": "B"
+        }
+      ]
+    },
+    {
+      "id": 3,
+      "title": "ClickHouse batch flushes (rate, by outcome)",
+      "type": "timeseries",
+      "datasource": { "type": "prometheus", "uid": "Prometheus" },
+      "gridPos": { "h": 8, "w": 12, "x": 0, "y": 8 },
+      "targets": [
+        {
+          "expr": "sum by (outcome) (rate(ingest_batch_flush_total[1m]))",
+          "legendFormat": "{{outcome}}",
+          "refId": "A"
+        }
+      ]
+    },
+    {
+      "id": 4,
+      "title": "ClickHouse batch flush duration (p95)",
+      "type": "timeseries",
+      "datasource": { "type": "prometheus", "uid": "Prometheus" },
+      "gridPos": { "h": 8, "w": 12, "x": 12, "y": 8 },
+      "fieldConfig": { "defaults": { "unit": "s" }, "overrides": [] },
+      "targets": [
+        {
+          "expr": "histogram_quantile(0.95, sum(rate(ingest_batch_flush_duration_seconds_bucket[5m])) by (le))",
+          "legendFormat": "p95",
+          "refId": "A"
+        }
+      ]
+    },
+    {
+      "id": 5,
+      "title": "RabbitMQ publishes (rate, by outcome)",
+      "type": "timeseries",
+      "datasource": { "type": "prometheus", "uid": "Prometheus" },
+      "gridPos": { "h": 8, "w": 12, "x": 0, "y": 16 },
+      "targets": [
+        {
+          "expr": "sum by (outcome) (rate(ingest_rabbitmq_publish_total[1m]))",
+          "legendFormat": "{{outcome}}",
+          "refId": "A"
+        }
+      ]
+    }
+  ]
+}

monitoring/grafana/provisioning/datasources/datasource.yml

@@ -0,0 +1,9 @@
+apiVersion: 1
+
+datasources:
+  - name: Prometheus
+    type: prometheus
+    access: proxy
+    url: http://prometheus:9090
+    isDefault: true
+    editable: false

monitoring/prometheus/prometheus.yml

@@ -0,0 +1,15 @@
+global:
+  scrape_interval: 15s
+
+scrape_configs:
+  - job_name: ingest-service
+    static_configs:
+      - targets: ["ingest-service:9101"]
+
+  - job_name: device-control-service
+    static_configs:
+      - targets: ["device-control-service:8090"]
+
+  - job_name: rule-engine-service
+    static_configs:
+      - targets: ["rule-engine-service:9102"]

services/device-control-service/README.md

@@ -49,6 +49,13 @@
   Laravel→device-control. gRPC-контракт между Go-сервисами (rule-engine)
   не тронут.
 
+## Метрики
+
+`GET /metrics` на том же порту, что и команды (`DEVICE_CONTROL_HTTP_PORT`)
+— не открывали отдельный порт ради этого. Счётчики/latency команд
+(TurnOn/TurnOff/SetLevel по action+outcome), MQTT-событий (telemetry/ack),
+переходов online/offline health-check.
+
 ## Запуск
 
 ```bash

services/device-control-service/cmd/device-control-service/main.go

@@ -16,12 +16,14 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"google.golang.org/grpc"
 
 	devicecontrol "git.cactoz.su/cacto/home_automatization/proto/device_control"
 	"git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/config"
 	"git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/healthcheck"
 	"git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/httpapi"
+	"git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/metrics"
 	"git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/mqttclient"
 	"git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/rabbitmq"
 	"git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/server"
@@ -74,6 +76,8 @@ func run(logger *slog.Logger) error {
 	}
 
 	checker := healthcheck.New(shadowStore, cfg.HealthCheckTimeout, cfg.HealthCheckInterval, func(deviceID string) {
+		metrics.HealthTransitionsTotal.WithLabelValues("offline").Inc()
+
 		pubCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
 		if err := publisher.PublishStatusChanged(pubCtx, deviceID, shadow.StatusOffline); err != nil {
@@ -93,10 +97,14 @@ func run(logger *slog.Logger) error {
 	devicecontrol.RegisterDeviceControlServer(grpcServer, srv)
 
 	// Same server instance, second transport — for callers where a full
-	// gRPC client is impractical (Laravel/PHP; see README).
+	// gRPC client is impractical (Laravel/PHP; see README). /metrics rides
+	// along on this same port rather than opening a third one.
+	mux := http.NewServeMux()
+	mux.Handle("/", httpapi.NewRouter(srv, logger))
+	mux.Handle("/metrics", promhttp.Handler())
 	httpServer := &http.Server{
 		Addr:    fmt.Sprintf(":%d", cfg.HTTPPort),
-		Handler: httpapi.NewRouter(srv, logger),
+		Handler: mux,
 	}
 
 	go func() {
@@ -145,8 +153,10 @@ func telemetryHandler(store *shadow.Store, publisher *rabbitmq.Publisher, logger
 		deviceID, ok := deviceIDFromTopic(topic)
 		if !ok {
 			logger.Warn("dropping telemetry from unparseable topic", "topic", topic)
+			metrics.MQTTEventsTotal.WithLabelValues("telemetry", "invalid").Inc()
 			return
 		}
+		metrics.MQTTEventsTotal.WithLabelValues("telemetry", "ok").Inc()
 		touch(context.Background(), store, publisher, deviceID, logger)
 	}
 }
@@ -160,14 +170,17 @@ func ackHandler(store *shadow.Store, publisher *rabbitmq.Publisher, logger *slog
 		deviceID, ok := deviceIDFromTopic(topic)
 		if !ok {
 			logger.Warn("dropping ack from unparseable topic", "topic", topic)
+			metrics.MQTTEventsTotal.WithLabelValues("ack", "invalid").Inc()
 			return
 		}
 
 		var ack ackPayload
 		if err := json.Unmarshal(payload, &ack); err != nil {
 			logger.Warn("dropping invalid ack payload", "device_id", deviceID, "error", err)
+			metrics.MQTTEventsTotal.WithLabelValues("ack", "invalid").Inc()
 			return
 		}
+		metrics.MQTTEventsTotal.WithLabelValues("ack", "ok").Inc()
 
 		ctx := context.Background()
 		if len(ack.State) > 0 {
@@ -186,6 +199,8 @@ func touch(ctx context.Context, store *shadow.Store, publisher *rabbitmq.Publish
 		return
 	}
 	if becameOnline {
+		metrics.HealthTransitionsTotal.WithLabelValues("online").Inc()
+
 		pubCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
 		defer cancel()
 		if err := publisher.PublishStatusChanged(pubCtx, deviceID, shadow.StatusOnline); err != nil {

services/device-control-service/go.mod

@@ -6,20 +6,27 @@ require (
 	git.cactoz.su/cacto/home_automatization/proto v0.0.0
 	github.com/alicebob/miniredis/v2 v2.38.0
 	github.com/eclipse/paho.mqtt.golang v1.5.1
+	github.com/prometheus/client_golang v1.24.1
 	github.com/rabbitmq/amqp091-go v1.13.0
 	github.com/redis/go-redis/v9 v9.21.0
 	google.golang.org/grpc v1.82.1
 )
 
 require (
+	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/gorilla/websocket v1.5.3 // indirect
+	github.com/kylelemons/godebug v1.1.0 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/prometheus/client_model v0.6.2 // indirect
+	github.com/prometheus/common v0.70.1 // indirect
+	github.com/prometheus/procfs v0.21.1 // indirect
 	github.com/yuin/gopher-lua v1.1.1 // indirect
 	go.uber.org/atomic v1.11.0 // indirect
-	golang.org/x/net v0.53.0 // indirect
+	golang.org/x/net v0.57.0 // indirect
-	golang.org/x/sync v0.20.0 // indirect
+	golang.org/x/sync v0.22.0 // indirect
-	golang.org/x/sys v0.43.0 // indirect
+	golang.org/x/sys v0.47.0 // indirect
-	golang.org/x/text v0.36.0 // indirect
+	golang.org/x/text v0.40.0 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 // indirect
 	google.golang.org/protobuf v1.36.11 // indirect
 )

services/device-control-service/go.sum

@@ -1,5 +1,7 @@
 github.com/alicebob/miniredis/v2 v2.38.0 h1:nZAzCR+Lj+Vxk4ZXzm2NuKq2O33RXj1XxJ2e2uP9jiw=
 github.com/alicebob/miniredis/v2 v2.38.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM=
+github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
+github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
 github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
 github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
@@ -22,16 +24,30 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
 github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk=
+github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
 github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE=
 github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
+github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
+github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU=
+github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE=
+github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
+github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
+github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY=
+github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc=
+github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI=
+github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY=
 github.com/rabbitmq/amqp091-go v1.13.0 h1:L8NA1WtF76C6KA3LAoufjfLgbist/If1UQYcsOjtxXA=
 github.com/rabbitmq/amqp091-go v1.13.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
 github.com/redis/go-redis/v9 v9.21.0 h1:FPBE4hhbAke+TLmcY3WkpbDffJEomdqPn3HYiqAtL9E=
 github.com/redis/go-redis/v9 v9.21.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA=
-github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
-github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M=
 github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw=
 github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs=
@@ -52,14 +68,16 @@ go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
 go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
-golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
+go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
-golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
+go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
-golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
+golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
-golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
+golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU=
-golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI=
+golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
-golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
-golang.org/x/text v0.36.0 h1:JfKh3XmcRPqZPKevfXVpI1wXPTqbkE5f7JA92a55Yxg=
+golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
-golang.org/x/text v0.36.0/go.mod h1:NIdBknypM8iqVmPiuco0Dh6P5Jcdk8lJL0CUebqK164=
+golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
+golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
 gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
 gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 h1:RmoJA1ujG+/lRGNfUnOMfhCy5EipVMyvUE+KNbPbTlw=
@@ -68,3 +86,5 @@ google.golang.org/grpc v1.82.1 h1:NnAxzGRA0677vCa4BUkOAnO5+FfQqVl9iUXeD0IqcGE=
 google.golang.org/grpc v1.82.1/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA=
 google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
 google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

services/device-control-service/internal/metrics/metrics.go

@@ -0,0 +1,32 @@
+// Package metrics defines device-control-service's Prometheus metrics.
+// Registered automatically (via promauto) into the default registry on
+// import; served alongside the command HTTP API's /metrics route.
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var (
+	CommandsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "device_control_commands_total",
+		Help: "Total number of TurnOn/TurnOff/SetLevel commands dispatched, by action and outcome.",
+	}, []string{"action", "outcome"})
+
+	CommandDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "device_control_command_duration_seconds",
+		Help:    "Duration of a command dispatch (Redis desired_state patch + MQTT publish), by action.",
+		Buckets: prometheus.DefBuckets,
+	}, []string{"action"})
+
+	MQTTEventsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "device_control_mqtt_events_total",
+		Help: "Total number of telemetry/ack messages received, by event type and outcome.",
+	}, []string{"event_type", "outcome"})
+
+	HealthTransitionsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "device_control_health_transitions_total",
+		Help: "Total number of device online/offline transitions detected.",
+	}, []string{"direction"})
+)

services/device-control-service/internal/server/server.go

@@ -9,8 +9,10 @@ import (
 	"encoding/json"
 	"fmt"
 	"log/slog"
+	"time"
 
 	devicecontrol "git.cactoz.su/cacto/home_automatization/proto/device_control"
+	"git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/metrics"
 )
 
 // ShadowPatcher is the subset of shadow.Store the server needs.
@@ -36,21 +38,21 @@ func New(shadow ShadowPatcher, mqtt CommandPublisher, logger *slog.Logger) *Serv
 }
 
 func (s *Server) TurnOn(ctx context.Context, req *devicecontrol.TurnOnRequest) (*devicecontrol.CommandResult, error) {
-	return s.dispatch(ctx, req.GetDeviceId(),
+	return s.dispatch(ctx, "turn_on", req.GetDeviceId(),
 		map[string]any{"power": "on"},
 		map[string]any{"action": "turn_on"},
 	)
 }
 
 func (s *Server) TurnOff(ctx context.Context, req *devicecontrol.TurnOffRequest) (*devicecontrol.CommandResult, error) {
-	return s.dispatch(ctx, req.GetDeviceId(),
+	return s.dispatch(ctx, "turn_off", req.GetDeviceId(),
 		map[string]any{"power": "off"},
 		map[string]any{"action": "turn_off"},
 	)
 }
 
 func (s *Server) SetLevel(ctx context.Context, req *devicecontrol.SetLevelRequest) (*devicecontrol.CommandResult, error) {
-	return s.dispatch(ctx, req.GetDeviceId(),
+	return s.dispatch(ctx, "set_level", req.GetDeviceId(),
 		map[string]any{"level": req.GetLevel()},
 		map[string]any{"action": "set_level", "level": req.GetLevel()},
 	)
@@ -60,27 +62,41 @@ func (s *Server) SetLevel(ctx context.Context, req *devicecontrol.SetLevelReques
 // gets an instant, optimistic view even if the device is offline), then hand
 // the command to MQTT. Business failures come back as CommandResult.Error,
 // not a gRPC error — the wire contract is success/error in one message.
-func (s *Server) dispatch(ctx context.Context, deviceID string, desiredPatch, commandPayload map[string]any) (*devicecontrol.CommandResult, error) {
+func (s *Server) dispatch(ctx context.Context, action, deviceID string, desiredPatch, commandPayload map[string]any) (*devicecontrol.CommandResult, error) {
+	start := time.Now()
+	result := s.doDispatch(ctx, action, deviceID, desiredPatch, commandPayload)
+
+	metrics.CommandDuration.WithLabelValues(action).Observe(time.Since(start).Seconds())
+	outcome := "success"
+	if !result.Success {
+		outcome = "error"
+	}
+	metrics.CommandsTotal.WithLabelValues(action, outcome).Inc()
+
+	return result, nil
+}
+
+func (s *Server) doDispatch(ctx context.Context, action, deviceID string, desiredPatch, commandPayload map[string]any) *devicecontrol.CommandResult {
 	if deviceID == "" {
-		return &devicecontrol.CommandResult{Success: false, Error: "device_id is required"}, nil
+		return &devicecontrol.CommandResult{Success: false, Error: "device_id is required"}
 	}
 
 	if err := s.shadow.PatchDesiredState(ctx, deviceID, desiredPatch); err != nil {
-		s.logger.Error("patch desired state failed", "device_id", deviceID, "error", err)
+		s.logger.Error("patch desired state failed", "action", action, "device_id", deviceID, "error", err)
-		return &devicecontrol.CommandResult{Success: false, Error: "failed to update desired state"}, nil
+		return &devicecontrol.CommandResult{Success: false, Error: "failed to update desired state"}
 	}
 
 	body, err := json.Marshal(commandPayload)
 	if err != nil {
-		s.logger.Error("encode command failed", "device_id", deviceID, "error", err)
+		s.logger.Error("encode command failed", "action", action, "device_id", deviceID, "error", err)
-		return &devicecontrol.CommandResult{Success: false, Error: "failed to encode command"}, nil
+		return &devicecontrol.CommandResult{Success: false, Error: "failed to encode command"}
 	}
 
 	topic := fmt.Sprintf("devices/%s/commands", deviceID)
 	if err := s.mqtt.Publish(topic, 1, false, body); err != nil {
-		s.logger.Error("publish command failed", "device_id", deviceID, "topic", topic, "error", err)
+		s.logger.Error("publish command failed", "action", action, "device_id", deviceID, "topic", topic, "error", err)
-		return &devicecontrol.CommandResult{Success: false, Error: "failed to dispatch command"}, nil
+		return &devicecontrol.CommandResult{Success: false, Error: "failed to dispatch command"}
 	}
 
-	return &devicecontrol.CommandResult{Success: true}, nil
+	return &devicecontrol.CommandResult{Success: true}
 }

services/device-control-service/internal/server/server_test.go

@@ -8,7 +8,10 @@ import (
 	"log/slog"
 	"testing"
 
+	"github.com/prometheus/client_golang/prometheus/testutil"
+
 	devicecontrol "git.cactoz.su/cacto/home_automatization/proto/device_control"
+	"git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/metrics"
 )
 
 type fakeShadow struct {
@@ -122,3 +125,27 @@ func TestDispatch_PublishFailure(t *testing.T) {
 		t.Fatal("expected failure when mqtt publish errors")
 	}
 }
+
+func TestDispatch_RecordsMetrics(t *testing.T) {
+	successCounter := metrics.CommandsTotal.WithLabelValues("turn_on", "success")
+	errorCounter := metrics.CommandsTotal.WithLabelValues("turn_on", "error")
+	before := testutil.ToFloat64(successCounter)
+
+	s := New(&fakeShadow{}, &fakeMQTT{}, discardLogger())
+	if _, err := s.TurnOn(context.Background(), &devicecontrol.TurnOnRequest{DeviceId: "d1"}); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if got := testutil.ToFloat64(successCounter); got != before+1 {
+		t.Fatalf("got turn_on/success counter %v, want %v", got, before+1)
+	}
+
+	beforeErr := testutil.ToFloat64(errorCounter)
+	s2 := New(&fakeShadow{err: errors.New("redis down")}, &fakeMQTT{}, discardLogger())
+	if _, err := s2.TurnOn(context.Background(), &devicecontrol.TurnOnRequest{DeviceId: "d1"}); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if got := testutil.ToFloat64(errorCounter); got != beforeErr+1 {
+		t.Fatalf("got turn_on/error counter %v, want %v", got, beforeErr+1)
+	}
+}

services/ingest-service/README.md

@@ -20,6 +20,13 @@ MQTT/ClickHouse/RabbitMQ, как и на диаграмме архитектур
 понадобится lookup через реестр устройств (с кэшированием, как rule-engine
 кэширует правила) — это естественное развитие, пока не реализовано.
 
+## Метрики
+
+`GET /metrics` (порт `INGEST_METRICS_PORT`, по умолчанию 9101) — формат
+Prometheus. Счётчики принятых/невалидных/отброшенных (backlog переполнен)
+MQTT-сообщений, флашей батча в ClickHouse (с latency и размером батча),
+публикаций в RabbitMQ.
+
 ## Запуск
 
 ```bash

services/ingest-service/cmd/ingest-service/main.go

@@ -5,16 +5,21 @@ package main
 
 import (
 	"context"
+	"fmt"
 	"log/slog"
+	"net/http"
 	"os"
 	"os/signal"
 	"sync"
 	"syscall"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+
 	"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/batch"
 	chstore "git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/clickhouse"
 	"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/config"
+	"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/metrics"
 	"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/mqttclient"
 	"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/rabbitmq"
 	"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/telemetry"
@@ -59,7 +64,7 @@ func run(logger *slog.Logger) error {
 	}
 	defer publisher.Close()
 
-	batcher := batch.New(cfg.BatchMaxSize, cfg.BatchFlushInterval, cfg.BatchFlushTimeout, store.InsertBatch, logger)
+	batcher := batch.New(cfg.BatchMaxSize, cfg.BatchFlushInterval, cfg.BatchFlushTimeout, instrumentedInsert(store), logger)
 	batcher.Start()
 	defer batcher.Stop()
 
@@ -77,6 +82,9 @@ func run(logger *slog.Logger) error {
 			cancel()
 			if err != nil {
 				logger.Error("publish reading event failed", "device_id", reading.DeviceID, "error", err)
+				metrics.RabbitMQPublishes.WithLabelValues("error").Inc()
+			} else {
+				metrics.RabbitMQPublishes.WithLabelValues("success").Inc()
 			}
 		}
 	}()
@@ -87,9 +95,12 @@ func run(logger *slog.Logger) error {
 		Topic:     cfg.MQTTTopic,
 		QoS:       1,
 	}, func(payload []byte, receivedAt time.Time) {
+		metrics.MQTTMessagesReceived.Inc()
+
 		reading, err := telemetry.ParseReading(payload, receivedAt)
 		if err != nil {
 			logger.Warn("dropping invalid telemetry payload", "error", err)
+			metrics.MQTTMessagesInvalid.Inc()
 			return
 		}
 
@@ -97,12 +108,24 @@ func run(logger *slog.Logger) error {
 		case incoming <- reading:
 		default:
 			logger.Error("dropping reading: worker backlog full", "device_id", reading.DeviceID)
+			metrics.MQTTMessagesDropped.Inc()
 		}
 	}, logger)
 	if err != nil {
 		return err
 	}
 
+	metricsServer := &http.Server{
+		Addr:    fmt.Sprintf(":%d", cfg.MetricsPort),
+		Handler: promhttp.Handler(),
+	}
+	go func() {
+		logger.Info("metrics server started", "metrics_port", cfg.MetricsPort)
+		if err := metricsServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			logger.Error("metrics server stopped", "error", err)
+		}
+	}()
+
 	logger.Info("ingest-service started", "mqtt_topic", cfg.MQTTTopic)
 
 	stop := make(chan os.Signal, 1)
@@ -114,5 +137,30 @@ func run(logger *slog.Logger) error {
 	close(incoming)
 	workerWG.Wait()
 
+	shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	if err := metricsServer.Shutdown(shutdownCtx); err != nil {
+		logger.Error("metrics server shutdown failed", "error", err)
+	}
+
 	return nil
 }
+
+// instrumentedInsert wraps store.InsertBatch with duration/size/outcome
+// metrics, without teaching the batch or clickhouse packages about Prometheus.
+func instrumentedInsert(store *chstore.Store) batch.FlushFunc {
+	return func(ctx context.Context, readings []telemetry.Reading) error {
+		start := time.Now()
+		err := store.InsertBatch(ctx, readings)
+
+		metrics.BatchFlushDuration.Observe(time.Since(start).Seconds())
+		metrics.BatchSize.Observe(float64(len(readings)))
+		if err != nil {
+			metrics.BatchFlushes.WithLabelValues("error").Inc()
+		} else {
+			metrics.BatchFlushes.WithLabelValues("success").Inc()
+		}
+
+		return err
+	}
+}

services/ingest-service/go.mod

@@ -5,26 +5,33 @@ go 1.25.0
 require (
 	github.com/ClickHouse/clickhouse-go/v2 v2.47.0
 	github.com/eclipse/paho.mqtt.golang v1.5.1
+	github.com/prometheus/client_golang v1.24.1
 	github.com/rabbitmq/amqp091-go v1.13.0
 )
 
 require (
 	github.com/ClickHouse/ch-go v0.73.0 // indirect
 	github.com/andybalholm/brotli v1.2.1 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/go-faster/city v1.0.1 // indirect
 	github.com/go-faster/errors v0.7.1 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/gorilla/websocket v1.5.3 // indirect
-	github.com/klauspost/compress v1.18.6 // indirect
+	github.com/klauspost/compress v1.19.1 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/paulmach/orb v0.13.0 // indirect
 	github.com/pierrec/lz4/v4 v4.1.27 // indirect
+	github.com/prometheus/client_model v0.6.2 // indirect
+	github.com/prometheus/common v0.70.1 // indirect
+	github.com/prometheus/procfs v0.21.1 // indirect
 	github.com/segmentio/asm v1.2.1 // indirect
 	github.com/shopspring/decimal v1.4.0 // indirect
 	go.opentelemetry.io/otel v1.44.0 // indirect
 	go.opentelemetry.io/otel/trace v1.44.0 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
-	golang.org/x/net v0.56.0 // indirect
+	golang.org/x/net v0.57.0 // indirect
-	golang.org/x/sync v0.20.0 // indirect
+	golang.org/x/sync v0.21.0 // indirect
-	golang.org/x/sys v0.46.0 // indirect
+	golang.org/x/sys v0.47.0 // indirect
+	google.golang.org/protobuf v1.36.11 // indirect
 )

services/ingest-service/go.sum

@@ -4,6 +4,8 @@ github.com/ClickHouse/clickhouse-go/v2 v2.47.0 h1:ZDAzrnKSOPTIsm4tdUNfrii2yc8dk4
 github.com/ClickHouse/clickhouse-go/v2 v2.47.0/go.mod h1:sPj7C7UYQ2MWHcfX+4eGN6nwnCqwUKfgO6PcwKpd6K8=
 github.com/andybalholm/brotli v1.2.1 h1:R+f5xP285VArJDRgowrfb9DqL18yVK0gKAW/F+eTWro=
 github.com/andybalholm/brotli v1.2.1/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
+github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
+github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
@@ -20,14 +22,26 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
 github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
-github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao=
+github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk=
-github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
+github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
+github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
+github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/paulmach/orb v0.13.0 h1:r7n7mQGGF+cj/CbcivEj9J3HGK+XR+yXnvzRdq9saIw=
 github.com/paulmach/orb v0.13.0/go.mod h1:6scRWINywA2Jf05dcjOfLfxrUIMECvTSG2MVbRLxu/k=
 github.com/pierrec/lz4/v4 v4.1.27 h1:+PhzhWDrjRj89TH2sw43nE3+4+W8lSxIuQadEHZyjUk=
 github.com/pierrec/lz4/v4 v4.1.27/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU=
+github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE=
+github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
+github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
+github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY=
+github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc=
+github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI=
+github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY=
 github.com/rabbitmq/amqp091-go v1.13.0 h1:L8NA1WtF76C6KA3LAoufjfLgbist/If1UQYcsOjtxXA=
 github.com/rabbitmq/amqp091-go v1.13.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
 github.com/segmentio/asm v1.2.1 h1:DTNbBqs57ioxAD4PrArqftgypG4/qNpXoJx8TVXxPR0=
@@ -44,14 +58,18 @@ go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/
 go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
+go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
 go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
 go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
-golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o=
+golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
-golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec=
+golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU=
-golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
+golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM=
-golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
+golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
-golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw=
+golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
-golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
+google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

services/ingest-service/internal/config/config.go

@@ -24,6 +24,8 @@ type Config struct {
 	BatchMaxSize       int
 	BatchFlushInterval time.Duration
 	BatchFlushTimeout  time.Duration
+
+	MetricsPort int
 }
 
 func Load() (Config, error) {
@@ -55,6 +57,9 @@ func Load() (Config, error) {
 	if cfg.BatchFlushTimeout, err = getEnvDuration("INGEST_BATCH_FLUSH_TIMEOUT", 10*time.Second); err != nil {
 		return Config{}, err
 	}
+	if cfg.MetricsPort, err = getEnvInt("INGEST_METRICS_PORT", 9101); err != nil {
+		return Config{}, err
+	}
 
 	return cfg, nil
 }

services/ingest-service/internal/metrics/metrics.go

@@ -0,0 +1,48 @@
+// Package metrics defines ingest-service's Prometheus metrics. Registered
+// automatically (via promauto) into the default registry on import; served
+// by main.go's dedicated metrics HTTP server.
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var (
+	MQTTMessagesReceived = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "ingest_mqtt_messages_received_total",
+		Help: "Total number of telemetry messages received from MQTT.",
+	})
+
+	MQTTMessagesInvalid = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "ingest_mqtt_messages_invalid_total",
+		Help: "Total number of telemetry messages dropped for failing validation.",
+	})
+
+	MQTTMessagesDropped = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "ingest_mqtt_messages_dropped_total",
+		Help: "Total number of readings dropped because the worker backlog was full.",
+	})
+
+	BatchFlushes = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "ingest_batch_flush_total",
+		Help: "Total number of batch flushes to ClickHouse, by outcome.",
+	}, []string{"outcome"})
+
+	BatchFlushDuration = promauto.NewHistogram(prometheus.HistogramOpts{
+		Name:    "ingest_batch_flush_duration_seconds",
+		Help:    "Duration of ClickHouse batch insert calls.",
+		Buckets: prometheus.DefBuckets,
+	})
+
+	BatchSize = promauto.NewHistogram(prometheus.HistogramOpts{
+		Name:    "ingest_batch_size",
+		Help:    "Number of readings per flushed batch.",
+		Buckets: []float64{1, 5, 10, 25, 50, 100, 250, 500, 1000},
+	})
+
+	RabbitMQPublishes = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "ingest_rabbitmq_publish_total",
+		Help: "Total number of new-reading events published to RabbitMQ, by outcome.",
+	}, []string{"outcome"})
+)

services/rule-engine-service/README.md

@@ -39,6 +39,12 @@
 - **Кэш правил — только для чтения активных правил**, никакой записи назад в
   Postgres. CRUD правил — это зона ответственности Laravel (этап 2).
 
+## Метрики
+
+`GET /metrics` (порт `RULE_ENGINE_METRICS_PORT`, по умолчанию 9102) —
+счётчики обработанных показаний, сработавших правил (по action_type и
+исходу), latency вызова device-control-service, обновлений кэша правил.
+
 ## Запуск
 
 ```bash

services/rule-engine-service/cmd/rule-engine-service/main.go

@@ -5,12 +5,16 @@ package main
 
 import (
 	"context"
+	"fmt"
 	"log/slog"
+	"net/http"
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 
 	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/config"
 	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/devicecontrolclient"
@@ -63,6 +67,17 @@ func run(logger *slog.Logger) error {
 
 	eng := engine.New(cache, dcClient, mq, logger)
 
+	metricsServer := &http.Server{
+		Addr:    fmt.Sprintf(":%d", cfg.MetricsPort),
+		Handler: promhttp.Handler(),
+	}
+	go func() {
+		logger.Info("metrics server started", "metrics_port", cfg.MetricsPort)
+		if err := metricsServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			logger.Error("metrics server stopped", "error", err)
+		}
+	}()
+
 	logger.Info("rule-engine-service started")
 
 	err = mq.ConsumeReadings(ctx, func(ctx context.Context, event rabbitmq.ReadingEvent) rabbitmq.HandleResult {
@@ -84,5 +99,12 @@ func run(logger *slog.Logger) error {
 	}
 
 	logger.Info("shutting down")
+
+	shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	if err := metricsServer.Shutdown(shutdownCtx); err != nil {
+		logger.Error("metrics server shutdown failed", "error", err)
+	}
+
 	return nil
 }

services/rule-engine-service/go.mod

@@ -5,18 +5,26 @@ go 1.25.0
 require (
 	git.cactoz.su/cacto/home_automatization/proto v0.0.0-00010101000000-000000000000
 	github.com/jackc/pgx/v5 v5.10.0
+	github.com/prometheus/client_golang v1.24.1
 	github.com/rabbitmq/amqp091-go v1.13.0
 	google.golang.org/grpc v1.82.1
 )
 
 require (
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
 	github.com/jackc/puddle/v2 v2.2.2 // indirect
-	golang.org/x/net v0.53.0 // indirect
+	github.com/kylelemons/godebug v1.1.0 // indirect
-	golang.org/x/sync v0.20.0 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
-	golang.org/x/sys v0.43.0 // indirect
+	github.com/prometheus/client_model v0.6.2 // indirect
-	golang.org/x/text v0.36.0 // indirect
+	github.com/prometheus/common v0.70.1 // indirect
+	github.com/prometheus/procfs v0.21.1 // indirect
+	golang.org/x/net v0.57.0 // indirect
+	golang.org/x/sync v0.22.0 // indirect
+	golang.org/x/sys v0.47.0 // indirect
+	golang.org/x/text v0.40.0 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 // indirect
 	google.golang.org/protobuf v1.36.11 // indirect
 )

services/rule-engine-service/go.sum

@@ -1,3 +1,5 @@
+github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
+github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -21,8 +23,22 @@ github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0=
 github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
 github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
 github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
+github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk=
+github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
+github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
+github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU=
+github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE=
+github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
+github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
+github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY=
+github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc=
+github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI=
+github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY=
 github.com/rabbitmq/amqp091-go v1.13.0 h1:L8NA1WtF76C6KA3LAoufjfLgbist/If1UQYcsOjtxXA=
 github.com/rabbitmq/amqp091-go v1.13.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
@@ -44,14 +60,16 @@ go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09
 go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
-golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
+go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
-golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
+go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
-golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
+golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
-golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
+golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU=
-golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI=
+golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
-golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
-golang.org/x/text v0.36.0 h1:JfKh3XmcRPqZPKevfXVpI1wXPTqbkE5f7JA92a55Yxg=
+golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
-golang.org/x/text v0.36.0/go.mod h1:NIdBknypM8iqVmPiuco0Dh6P5Jcdk8lJL0CUebqK164=
+golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
+golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
 gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
 gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 h1:RmoJA1ujG+/lRGNfUnOMfhCy5EipVMyvUE+KNbPbTlw=

services/rule-engine-service/internal/config/config.go

@@ -5,6 +5,7 @@ package config
 import (
 	"fmt"
 	"os"
+	"strconv"
 	"time"
 )
 
@@ -16,6 +17,8 @@ type Config struct {
 	RabbitMQURL string
 
 	RuleCacheRefreshInterval time.Duration
+
+	MetricsPort int
 }
 
 func Load() (Config, error) {
@@ -47,6 +50,10 @@ func Load() (Config, error) {
 	}
 	cfg.RuleCacheRefreshInterval = interval
 
+	if cfg.MetricsPort, err = getEnvInt("RULE_ENGINE_METRICS_PORT", 9102); err != nil {
+		return Config{}, err
+	}
+
 	return cfg, nil
 }
 
@@ -57,6 +64,18 @@ func getEnv(key, fallback string) string {
 	return fallback
 }
 
+func getEnvInt(key string, fallback int) (int, error) {
+	v := os.Getenv(key)
+	if v == "" {
+		return fallback, nil
+	}
+	n, err := strconv.Atoi(v)
+	if err != nil {
+		return 0, fmt.Errorf("%s: %w", key, err)
+	}
+	return n, nil
+}
+
 func getEnvDuration(key string, fallback time.Duration) (time.Duration, error) {
 	v := os.Getenv(key)
 	if v == "" {

services/rule-engine-service/internal/engine/engine.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/devicecontrolclient"
+	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/metrics"
 	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/rules"
 )
 
@@ -57,6 +58,8 @@ func New(ruleSource RuleSource, dispatcher Dispatcher, publisher TriggerPublishe
 // the whole reading for. A rule the device itself rejected (bad device_id,
 // unsupported action) is terminal and does not cause a retry.
 func (e *Engine) HandleReading(ctx context.Context, r Reading) error {
+	metrics.ReadingsConsumed.Inc()
+
 	matched := e.rules.MatchingRules(r.DeviceID, r.SensorType)
 
 	var firstTransportErr error
@@ -69,8 +72,11 @@ func (e *Engine) HandleReading(ctx context.Context, r Reading) error {
 		if !matchesCondition {
 			continue
 		}
+		metrics.RulesMatched.Inc()
 
+		start := time.Now()
 		result, dispatchErr := e.dispatcher.Dispatch(ctx, rule.TargetDeviceID, rule.ActionType, rule.ActionParams)
+		metrics.DispatchDuration.WithLabelValues(rule.ActionType).Observe(time.Since(start).Seconds())
 
 		success := dispatchErr == nil && result.Success
 		errMsg := result.Error
@@ -78,13 +84,16 @@ func (e *Engine) HandleReading(ctx context.Context, r Reading) error {
 		case dispatchErr != nil:
 			errMsg = dispatchErr.Error()
 			e.logger.Error("dispatch action failed", "rule_id", rule.ID, "target_device_id", rule.TargetDeviceID, "error", dispatchErr)
+			metrics.RulesTriggered.WithLabelValues(rule.ActionType, "transport_error").Inc()
 			if firstTransportErr == nil {
 				firstTransportErr = dispatchErr
 			}
 		case !result.Success:
 			e.logger.Warn("device rejected command", "rule_id", rule.ID, "target_device_id", rule.TargetDeviceID, "error", result.Error)
+			metrics.RulesTriggered.WithLabelValues(rule.ActionType, "rejected").Inc()
 		default:
 			e.logger.Info("rule triggered", "rule_id", rule.ID, "target_device_id", rule.TargetDeviceID, "action_type", rule.ActionType)
+			metrics.RulesTriggered.WithLabelValues(rule.ActionType, "success").Inc()
 		}
 
 		if pubErr := e.publisher.PublishRuleTriggered(ctx, rule.ID, rule.ZoneID, rule.TargetDeviceID, rule.ActionType, success, errMsg); pubErr != nil {

services/rule-engine-service/internal/engine/engine_test.go

@@ -7,7 +7,10 @@ import (
 	"log/slog"
 	"testing"
 
+	"github.com/prometheus/client_golang/prometheus/testutil"
+
 	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/devicecontrolclient"
+	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/metrics"
 	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/rules"
 )
 
@@ -154,3 +157,30 @@ func TestHandleReading_InvalidOperatorSkipsRuleButContinues(t *testing.T) {
 		t.Fatalf("got dispatch calls %+v, want only the valid rule dispatched", dispatcher.calls)
 	}
 }
+
+func TestHandleReading_RecordsMetrics(t *testing.T) {
+	source := &fakeRuleSource{rules: []rules.Rule{
+		{ID: 1, ConditionOperator: ">", ConditionValue: 28, TargetDeviceID: "fan-1", ActionType: "turn_on"},
+	}}
+	dispatcher := &fakeDispatcher{result: devicecontrolclient.Result{Success: true}}
+	e := New(source, dispatcher, &fakePublisher{}, discardLogger())
+
+	triggeredCounter := metrics.RulesTriggered.WithLabelValues("turn_on", "success")
+	beforeConsumed := testutil.ToFloat64(metrics.ReadingsConsumed)
+	beforeMatched := testutil.ToFloat64(metrics.RulesMatched)
+	beforeTriggered := testutil.ToFloat64(triggeredCounter)
+
+	if err := e.HandleReading(context.Background(), Reading{Value: 30}); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if got := testutil.ToFloat64(metrics.ReadingsConsumed); got != beforeConsumed+1 {
+		t.Fatalf("got readings_consumed %v, want %v", got, beforeConsumed+1)
+	}
+	if got := testutil.ToFloat64(metrics.RulesMatched); got != beforeMatched+1 {
+		t.Fatalf("got rules_matched %v, want %v", got, beforeMatched+1)
+	}
+	if got := testutil.ToFloat64(triggeredCounter); got != beforeTriggered+1 {
+		t.Fatalf("got rules_triggered{turn_on,success} %v, want %v", got, beforeTriggered+1)
+	}
+}

services/rule-engine-service/internal/metrics/metrics.go

@@ -0,0 +1,37 @@
+// Package metrics defines rule-engine-service's Prometheus metrics.
+// Registered automatically (via promauto) into the default registry on
+// import; served by main.go's dedicated metrics HTTP server.
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var (
+	ReadingsConsumed = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "rule_engine_readings_consumed_total",
+		Help: "Total number of telemetry.new_reading events consumed from RabbitMQ.",
+	})
+
+	RulesMatched = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "rule_engine_rules_matched_total",
+		Help: "Total number of rule conditions that evaluated true and were dispatched.",
+	})
+
+	RulesTriggered = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "rule_engine_rules_triggered_total",
+		Help: "Total number of rule dispatch attempts, by action type and outcome.",
+	}, []string{"action_type", "outcome"})
+
+	DispatchDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "rule_engine_dispatch_duration_seconds",
+		Help:    "Duration of the gRPC/HTTP call to device-control-service, by action type.",
+		Buckets: prometheus.DefBuckets,
+	}, []string{"action_type"})
+
+	CacheRefreshes = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "rule_engine_cache_refresh_total",
+		Help: "Total number of automation_rules cache refreshes from PostgreSQL, by outcome.",
+	}, []string{"outcome"})
+)

services/rule-engine-service/internal/rulecache/cache.go

@@ -10,6 +10,7 @@ import (
 	"sync"
 	"time"
 
+	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/metrics"
 	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/rules"
 )
 
@@ -88,12 +89,14 @@ func (c *Cache) loop() {
 func (c *Cache) refresh(ctx context.Context) error {
 	dbRows, err := c.fetcher.FetchActiveRules(ctx)
 	if err != nil {
+		metrics.CacheRefreshes.WithLabelValues("error").Inc()
 		return err
 	}
 	index := buildIndex(dbRows, c.logger)
 	c.mu.Lock()
 	c.index = index
 	c.mu.Unlock()
+	metrics.CacheRefreshes.WithLabelValues("success").Inc()
 	return nil
 }