4 Commits

5 changed files with 81 additions and 26 deletions

View File

@@ -1,4 +1,5 @@
export const cloudIngestUrl = process.env.CLOUD_INGEST_URL as string
export const demoCloudIngestUrl = process.env.DEMO_CLOUD_INGEST_URL as string
export function postCloudIngest(body: unknown): Promise<Response> {
return fetch(cloudIngestUrl, {
@@ -10,3 +11,14 @@ export function postCloudIngest(body: unknown): Promise<Response> {
body: JSON.stringify(body),
})
}
export function postDemoCloudIngest(body: unknown): Promise<Response> {
return fetch(demoCloudIngestUrl, {
method: 'POST',
headers: {
'content-type': 'application/json',
'x-api-key': process.env.CLOUD_INGEST_API_KEY as string,
},
body: JSON.stringify(body),
})
}

View File

@@ -1,4 +1,4 @@
import { postCloudIngest } from '../cloud-ingest.js'
import { postCloudIngest, postDemoCloudIngest } from '../cloud-ingest.js'
import { log } from '../helpers/log.js'
import type { TopicHandler } from '../types.js'
@@ -36,10 +36,10 @@ type Edge5ModbusV3Metric = {
edge: string
timestamp: string
tag: string
value: number
value: number | null
}
function toFloatCDAB(reg0: number, reg1: number): number {
function toFloatCDAB(reg0: number, reg1: number): number | null {
const buf = Buffer.allocUnsafe(4)
buf.writeUInt16BE(reg1, 0)
buf.writeUInt16BE(reg0, 2)
@@ -63,16 +63,7 @@ function toMetrics(values: number[], timestamp: string): Edge5ModbusV3Metric[] {
for (let i = 0; i < values.length; i += 2) {
const tag = TAGS[i / 2]
const value = toFloatCDAB(values[i], values[i + 1])
if (isNaN(value)) {
if (CLOUD_INGEST_TAGS.has(tag)) {
log(
`edge5.modbus.v3: не удалось собрать число для ${tag}, ` +
`исходные values[${i}]=${values[i]}, values[${i + 1}]=${values[i + 1]}, пропуск`,
)
}
continue
}
const value = toFloatCDAB(values[i], values[i + 1]); // null is possible to send
metrics.push({
edge: EDGE,
@@ -97,6 +88,9 @@ async function postMetrics(metrics: Edge5ModbusV3Metric[]): Promise<void> {
const text = await response.text()
throw new Error(text || `${response.status} ${response.statusText}`)
}
void postDemoCloudIngest(metric); // для демо-системы
} catch (error) {
const message = error instanceof Error ? error.message : String(error)
errors.push(new Error(`${metric.tag}: ${message}`))
@@ -104,7 +98,7 @@ async function postMetrics(metrics: Edge5ModbusV3Metric[]): Promise<void> {
}
}
log('edge5.modbus.v3.posted', { total: toPost.length });
log(`edge5.modbus.v3.data ↑ ${toPost.length} tags`);
if (errors.length) {
throw new Error(`Failed to post ${errors.length}/${toPost.length} edge5 modbus v3 metrics`)
@@ -113,8 +107,6 @@ async function postMetrics(metrics: Edge5ModbusV3Metric[]): Promise<void> {
export const handleEdge5ModbusV3: TopicHandler = async (topic, payload) => {
const timestamp = new Date().toISOString();
log('edge5.modbus.v3.received', { topic, bytes: payload.length, timestamp })
const values = parseValues(payload);
const metrics = toMetrics(values, timestamp);
await postMetrics(metrics);

View File

@@ -9,10 +9,10 @@ type Edge5iiModbusMetric = {
edge: string
timestamp: string
tag: string
value: number
value: number | null
};
function toFloatCDAB(reg0: number, reg1: number): number {
function toFloatCDAB(reg0: number, reg1: number): number | null {
const buf = Buffer.allocUnsafe(4)
buf.writeUInt16BE(reg1, 0)
buf.writeUInt16BE(reg0, 2)
@@ -36,14 +36,7 @@ function toMetrics(values: number[], timestamp: string): Edge5iiModbusMetric[] {
for (let i = 0; i < values.length; i += 2) {
const tag = `register-${String(i).padStart(3, '0')}`
const value = toFloatCDAB(values[i], values[i + 1])
if (isNaN(value)) {
log(
`${EDGE}.modbus: не удалось собрать число для ${tag}, ` +
`исходные values[${i}]=${values[i]}, values[${i + 1}]=${values[i + 1]}, пропуск`,
)
continue
};
const value = toFloatCDAB(values[i], values[i + 1]); // null is possible to send
metrics.push({
edge: EDGE,

View File

@@ -1,3 +1,4 @@
import './process-errors.js'
import './handlers/index.js'
import './http/server.js'
import './mqtt.js'

57
app/src/process-errors.ts Normal file
View File

@@ -0,0 +1,57 @@
type ErrorWithCause = Error & {
cause?: unknown
}
type NodeError = Error & {
code?: string
address?: string
port?: number
}
// Undici кладёт системную ошибку соединения внутрь error.cause.
function getFetchCause(error: unknown): NodeError | null {
if (!(error instanceof Error)) {
return null
}
const cause = (error as ErrorWithCause).cause
return cause instanceof Error ? (cause as NodeError) : null
}
// Обрабатываем только подсвеченный кейс: backend отказал в TCP-соединении.
function isFetchConnectionRefused(error: unknown): boolean {
if (!(error instanceof Error)) {
return false
}
const cause = getFetchCause(error)
return error.message === 'fetch failed' && cause?.code === 'ECONNREFUSED'
}
// Логируем короткую диагностическую форму без огромного stack trace.
function formatFetchError(error: unknown): object {
if (!(error instanceof Error)) {
return { message: String(error) }
}
const cause = getFetchCause(error)
return {
message: error.message,
cause: cause?.message,
code: cause?.code,
address: cause?.address,
port: cause?.port,
}
}
// Последний рубеж защиты для fire-and-forget fetch-запросов.
// Если backend недоступен, Node.js 22 может превратить незахваченный
// fetch failed / ECONNREFUSED в падение всего процесса mqtt-worker.
process.on('unhandledRejection', (error) => {
if (isFetchConnectionRefused(error)) {
console.error('process.unhandled_fetch_refused', formatFetchError(error))
return
}
console.error('process.unhandled_rejection', error)
})