61dc0d60e1
* Add linting rule to sort imports * remove spacing between import groups * changeset * changeset * prettier config fine tuning * forbid use of deprecated imports * tidy
119 lines
3.2 KiB
JavaScript
119 lines
3.2 KiB
JavaScript
/* eslint-env node */
|
|
/* eslint-disable import/order */
|
|
|
|
'use strict'
|
|
|
|
const dd = require('dd-trace')
|
|
|
|
dd.tracer
|
|
.init()
|
|
.use('http2', {
|
|
client: true, // calls into dataplane
|
|
server: false,
|
|
})
|
|
.use('express', {
|
|
hooks: {
|
|
request: (span, req) => {
|
|
maintainXrpcResource(span, req)
|
|
},
|
|
},
|
|
})
|
|
|
|
// modify tracer in order to track calls to dataplane as a service with proper resource names
|
|
const DATAPLANE_PREFIX = '/bsky.Service/'
|
|
const origStartSpan = dd.tracer._tracer.startSpan
|
|
dd.tracer._tracer.startSpan = function (name, options) {
|
|
if (
|
|
name !== 'http.request' ||
|
|
options?.tags?.component !== 'http2' ||
|
|
!options?.tags?.['http.url']
|
|
) {
|
|
return origStartSpan.call(this, name, options)
|
|
}
|
|
const uri = new URL(options.tags['http.url'])
|
|
if (!uri.pathname.startsWith(DATAPLANE_PREFIX)) {
|
|
return origStartSpan.call(this, name, options)
|
|
}
|
|
options.tags['service.name'] = 'dataplane-bsky'
|
|
options.tags['resource.name'] = uri.pathname.slice(DATAPLANE_PREFIX.length)
|
|
return origStartSpan.call(this, name, options)
|
|
}
|
|
|
|
// Tracer code above must come before anything else
|
|
const assert = require('node:assert')
|
|
const cluster = require('node:cluster')
|
|
const path = require('node:path')
|
|
|
|
const { BskyAppView, ServerConfig } = require('@atproto/bsky')
|
|
const { Secp256k1Keypair } = require('@atproto/crypto')
|
|
|
|
const main = async () => {
|
|
const env = getEnv()
|
|
const config = ServerConfig.readEnv()
|
|
assert(env.serviceSigningKey, 'must set BSKY_SERVICE_SIGNING_KEY')
|
|
const signingKey = await Secp256k1Keypair.import(env.serviceSigningKey)
|
|
const bsky = BskyAppView.create({ config, signingKey })
|
|
await bsky.start()
|
|
// Graceful shutdown (see also https://aws.amazon.com/blogs/containers/graceful-shutdowns-with-ecs/)
|
|
const shutdown = async () => {
|
|
await bsky.destroy()
|
|
}
|
|
process.on('SIGTERM', shutdown)
|
|
process.on('disconnect', shutdown) // when clustering
|
|
}
|
|
|
|
const getEnv = () => ({
|
|
serviceSigningKey: process.env.BSKY_SERVICE_SIGNING_KEY || undefined,
|
|
})
|
|
|
|
const maybeParseInt = (str) => {
|
|
if (!str) return
|
|
const int = parseInt(str, 10)
|
|
if (isNaN(int)) return
|
|
return int
|
|
}
|
|
|
|
const maintainXrpcResource = (span, req) => {
|
|
// Show actual xrpc method as resource rather than the route pattern
|
|
if (span && req.originalUrl?.startsWith('/xrpc/')) {
|
|
span.setTag(
|
|
'resource.name',
|
|
[
|
|
req.method,
|
|
path.posix.join(req.baseUrl || '', req.path || '', '/').slice(0, -1), // Ensures no trailing slash
|
|
]
|
|
.filter(Boolean)
|
|
.join(' '),
|
|
)
|
|
}
|
|
}
|
|
|
|
const workerCount = maybeParseInt(process.env.CLUSTER_WORKER_COUNT)
|
|
|
|
if (workerCount) {
|
|
if (cluster.isPrimary) {
|
|
console.log(`primary ${process.pid} is running`)
|
|
const workers = new Set()
|
|
for (let i = 0; i < workerCount; ++i) {
|
|
workers.add(cluster.fork())
|
|
}
|
|
let teardown = false
|
|
cluster.on('exit', (worker) => {
|
|
workers.delete(worker)
|
|
if (!teardown) {
|
|
workers.add(cluster.fork()) // restart on crash
|
|
}
|
|
})
|
|
process.on('SIGTERM', () => {
|
|
teardown = true
|
|
console.log('disconnecting workers')
|
|
workers.forEach((w) => w.disconnect())
|
|
})
|
|
} else {
|
|
console.log(`worker ${process.pid} is running`)
|
|
main()
|
|
}
|
|
} else {
|
|
main() // non-clustering
|
|
}
|