This repository has no description
6.3 kB
215 lines
1# Vector on every machine: host metrics and the journal, written to every
2# monitoring server (dual write). Each sink has its own disk buffer, so a
3# server outage or a laptop being offline doesn't lose data.
4{
5 config,
6 lib,
7 pkgs,
8 ...
9}:
10let
11 inherit (import ./default.nix) servers ports;
12
13 # Containers have no disks of their own and lack the capabilities the
14 # SMART exporter needs.
15 smart = !config.boot.isContainer;
16
17 buffer = {
18 type = "disk";
19 max_size = 512 * 1024 * 1024;
20 when_full = "drop_newest";
21 };
22
23 sinksFor = server: {
24 "metrics_${server}" = {
25 type = "prometheus_remote_write";
26 inputs = [
27 "host_metrics"
28 "revision_metric"
29 "memory_pressure_metric"
30 ];
31 endpoint = "http://${server}.r:${toString ports.victoriametrics}/api/v1/write";
32 healthcheck.enabled = false;
33 inherit buffer;
34 };
35 "logs_${server}" = {
36 type = "elasticsearch";
37 inputs = [ "journal" ];
38 endpoints = [ "http://${server}.r:${toString ports.victorialogs}/insert/elasticsearch/" ];
39 mode = "bulk";
40 api_version = "v8";
41 healthcheck.enabled = false;
42 query = {
43 _msg_field = "message";
44 _time_field = "timestamp";
45 _stream_fields = "host,unit";
46 };
47 inherit buffer;
48 };
49 };
50in
51{
52 services.prometheus.exporters.smartctl = lib.mkIf smart {
53 enable = true;
54 listenAddress = "127.0.0.1";
55 };
56
57 services.vector = {
58 enable = true;
59 journaldAccess = true;
60 settings = {
61 data_dir = "/var/lib/vector";
62
63 sources.host_metrics_raw = {
64 type = "host_metrics";
65 collectors = [
66 "cpu"
67 "memory"
68 "disk"
69 "filesystem"
70 "load"
71 "network"
72 "host"
73 ];
74 filesystem.filesystems.excludes = [
75 "tmpfs"
76 "devtmpfs"
77 "overlay"
78 "squashfs"
79 "ramfs"
80 ];
81 scrape_interval_secs = 30;
82 };
83 sources.internal_metrics_raw = {
84 type = "internal_metrics";
85 scrape_interval_secs = 30;
86 };
87 sources.journald_raw.type = "journald";
88 sources.smartctl_raw = lib.mkIf smart {
89 type = "prometheus_scrape";
90 endpoints = [
91 "http://127.0.0.1:${toString config.services.prometheus.exporters.smartctl.port}/metrics"
92 ];
93 scrape_interval_secs = 60;
94 };
95
96 # Memory PSI: the time some task stalled on memory, as a counter in
97 # seconds, so rate() over it is the stalled fraction of a window.
98 sources.memory_pressure_raw = {
99 type = "exec";
100 mode = "scheduled";
101 scheduled.exec_interval_secs = 30;
102 command = [
103 "${pkgs.coreutils}/bin/cat"
104 "/proc/pressure/memory"
105 ];
106 include_stderr = false;
107 };
108 transforms.memory_pressure = {
109 type = "remap";
110 inputs = [ "memory_pressure_raw" ];
111 source = ''
112 fields = parse_key_value!(.message)
113 if !exists(fields.some) { abort }
114 . = { "seconds": to_int!(fields.total) / 1000000 }
115 '';
116 drop_on_abort = true;
117 };
118 transforms.memory_pressure_metric = {
119 type = "log_to_metric";
120 inputs = [ "memory_pressure" ];
121 metrics = [
122 {
123 type = "counter";
124 kind = "absolute";
125 increment_by_value = true;
126 field = "seconds";
127 name = "host_pressure_memory_some_seconds_total";
128 tags.host = config.networking.hostName;
129 }
130 ];
131 };
132
133 # The superconfig revision the machine runs, written by deploy and
134 # autoUpgrade after a switch. Read at runtime: it is in no derivation,
135 # so a new commit doesn't change every machine's closure.
136 sources.revision_raw = {
137 type = "exec";
138 mode = "scheduled";
139 scheduled.exec_interval_secs = 30;
140 command = [
141 pkgs.runtimeShell
142 "-c"
143 "cat /var/lib/superconfig-revision 2>/dev/null || echo unknown"
144 ];
145 include_stderr = false;
146 };
147 transforms.revision = {
148 type = "remap";
149 inputs = [ "revision_raw" ];
150 source = ''
151 rev = strip_whitespace(string!(.message))
152 if rev == "" { rev = "unknown" }
153 . = {
154 "value": 1,
155 "revision": rev,
156 "revision_short": replace(rev, r'^([0-9a-f]{7})[0-9a-f]{33}', "$1"),
157 }
158 '';
159 };
160 transforms.revision_metric = {
161 type = "log_to_metric";
162 inputs = [ "revision" ];
163 metrics = [
164 {
165 type = "gauge";
166 field = "value";
167 name = "nixos_system_info";
168 tags = {
169 host = config.networking.hostName;
170 revision = "{{revision}}";
171 revision_short = "{{revision_short}}";
172 };
173 }
174 ];
175 };
176
177 # The NixOS hostname as host label on every metric, the same value the
178 # journal events get below (alerts and dashboards match on it).
179 transforms.host_metrics = {
180 type = "remap";
181 inputs = [
182 "host_metrics_raw"
183 "internal_metrics_raw"
184 ]
185 ++ lib.optional smart "smartctl_raw";
186 source = ''
187 .tags.host = "${config.networking.hostName}"
188 '';
189 };
190
191 # Keep only what we query on: drop debug messages and the many
192 # journald bookkeeping fields. unit prefers UNIT/USER_UNIT, which
193 # systemd sets on its own messages about a unit ("Failed with result"),
194 # so those land in the stream of the unit they are about.
195 transforms.journal = {
196 type = "remap";
197 inputs = [ "journald_raw" ];
198 source = ''
199 if .PRIORITY == "7" { abort }
200 . = {
201 "timestamp": .timestamp,
202 "message": .message,
203 "host": "${config.networking.hostName}",
204 "unit": .UNIT || .USER_UNIT || ._SYSTEMD_USER_UNIT || ._SYSTEMD_UNIT || .SYSLOG_IDENTIFIER || "unknown",
205 "priority": .PRIORITY,
206 "identifier": .SYSLOG_IDENTIFIER,
207 }
208 '';
209 drop_on_abort = true;
210 };
211
212 sinks = lib.mergeAttrsList (map sinksFor servers);
213 };
214 };
215}