feat(xo-server): multi processes

This commit is contained in:
Julien Fontanet 2018-03-19 12:10:12 +01:00
parent 0e2270fb6e
commit 766175b4a0
24 changed files with 1334 additions and 698 deletions

View File

@ -0,0 +1,3 @@
module.exports = require('../../@xen-orchestra/babel-config')(
require('./package.json')
)

View File

@ -0,0 +1,24 @@
/benchmark/
/benchmarks/
*.bench.js
*.bench.js.map
/examples/
example.js
example.js.map
*.example.js
*.example.js.map
/fixture/
/fixtures/
*.fixture.js
*.fixture.js.map
*.fixtures.js
*.fixtures.js.map
/test/
/tests/
*.spec.js
*.spec.js.map
__snapshots__/

View File

@ -0,0 +1,149 @@
# @xen-orchestra/log [![Build Status](https://travis-ci.org/vatesfr/xen-orchestra.png?branch=master)](https://travis-ci.org/vatesfr/xen-orchestra)
> ${pkg.description}
## Install
Installation of the [npm package](https://npmjs.org/package/@xen-orchestra/log):
```
> npm install --save @xen-orchestra/log
```
## Usage
Everywhere something should be logged:
```js
import { createLogger } from '@xen-orchestra/log'
const log = createLogger('xo-server-api')
log.warn('foo')
```
Then at application level you can choose how to handle these logs:
```js
import configure from '@xen-orchestra/log/configure'
import createConsoleTransport from '@xen-orchestra/log/transports/console'
import createEmailTransport from '@xen-orchestra/log/transports/email'
configure([
{
// if filter is a string, then it is pattern
// (https://github.com/visionmedia/debug#wildcards) which is
// matched against the namespace of the logs
filter: process.env.DEBUG,
transport: createConsoleTransport()
},
{
// only levels >= warn
level: 'warn',
transport: createEmaileTransport({
service: 'gmail',
auth: {
user: 'jane.smith@gmail.com',
pass: 'H&NbECcpXF|pyXe#%ZEb'
},
from: 'jane.smith@gmail.com',
to: [
'jane.smith@gmail.com',
'sam.doe@yahoo.com'
]
})
}
])
```
### Transports
#### Console
```js
import createConsoleTransport from '@xen-orchestra/log/transports/console'
configure(createConsoleTransport())
```
#### Email
Optional dependency:
```
> yarn add nodemailer pretty-format
```
Configuration:
```js
import createEmailTransport from '@xen-orchestra/log/transports/email'
configure(createEmailTransport({
service: 'gmail',
auth: {
user: 'jane.smith@gmail.com',
pass: 'H&NbECcpXF|pyXe#%ZEb'
},
from: 'jane.smith@gmail.com',
to: [
'jane.smith@gmail.com',
'sam.doe@yahoo.com'
]
}))
```
#### Syslog
Optional dependency:
```
> yarn add split-host syslog-client
```
Configuration:
```js
import createSyslogTransport from '@xen-orchestra/log/transports/syslog'
// By default, log to udp://localhost:514
configure(createSyslogTransport())
// But TCP, a different host, or a different port can be used
configure(createSyslogTransport('tcp://syslog.company.lan'))
```
## Development
```
# Install dependencies
> yarn
# Run the tests
> yarn test
# Continuously compile
> yarn dev
# Continuously run the tests
> yarn dev-test
# Build for production (automatically called by npm install)
> yarn build
```
## Contributions
Contributions are *very* welcomed, either on the documentation or on
the code.
You may:
- report any [issue](https://github.com/vatesfr/xo-web/issues/)
you've encountered;
- fork and create a pull request.
## License
ISC © [Vates SAS](https://vates.fr)

View File

@ -0,0 +1 @@
module.exports = require('./dist/configure')

View File

@ -0,0 +1,52 @@
{
"private": true,
"name": "@xen-orchestra/log",
"version": "0.0.0",
"license": "ISC",
"description": "",
"keywords": [],
"homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/packages/@xen-orchestra/log",
"bugs": "https://github.com/vatesfr/xen-orchestra/issues",
"repository": {
"type": "git",
"url": "https://github.com/vatesfr/xen-orchestra.git"
},
"author": {
"name": "Julien Fontanet",
"email": "julien.fontanet@vates.fr"
},
"preferGlobal": false,
"main": "dist/",
"bin": {},
"files": [
"dist/"
],
"browserslist": [
">2%"
],
"engines": {
"node": ">=4"
},
"dependencies": {
"@babel/polyfill": "7.0.0-beta.42",
"lodash": "^4.17.4",
"promise-toolbox": "^0.9.5"
},
"devDependencies": {
"@babel/cli": "7.0.0-beta.42",
"@babel/core": "7.0.0-beta.42",
"@babel/preset-env": "7.0.0-beta.42",
"@babel/preset-flow": "7.0.0-beta.42",
"babel-plugin-lodash": "^3.3.2",
"cross-env": "^5.1.3",
"rimraf": "^2.6.2"
},
"scripts": {
"build": "cross-env NODE_ENV=production babel --source-maps --out-dir=dist/ src/",
"clean": "rimraf dist/",
"dev": "cross-env NODE_ENV=development babel --watch --source-maps --out-dir=dist/ src/",
"prebuild": "yarn run clean",
"predev": "yarn run prebuild",
"prepublishOnly": "yarn run build"
}
}

View File

@ -0,0 +1,105 @@
import createConsoleTransport from './transports/console'
import LEVELS, { resolve } from './levels'
import { compileGlobPattern } from './utils'
// ===================================================================
const createTransport = config => {
if (typeof config === 'function') {
return config
}
if (Array.isArray(config)) {
const transports = config.map(createTransport)
const { length } = transports
return function () {
for (let i = 0; i < length; ++i) {
transports[i].apply(this, arguments)
}
}
}
let { filter, transport } = config
const level = resolve(config.level)
if (filter !== undefined) {
if (typeof filter === 'string') {
const re = compileGlobPattern(filter)
filter = log => re.test(log.namespace)
}
const orig = transport
transport = function (log) {
if ((level !== undefined && log.level >= level) || filter(log)) {
return orig.apply(this, arguments)
}
}
} else if (level !== undefined) {
const orig = transport
transport = function (log) {
if (log.level >= level) {
return orig.apply(this, arguments)
}
}
}
return transport
}
let transport = createTransport({
// display warnings or above, and all that are enabled via DEBUG or
// NODE_DEBUG env
filter: process.env.DEBUG || process.env.NODE_DEBUG,
level: LEVELS.INFO,
transport: createConsoleTransport(),
})
const symbol =
typeof Symbol !== 'undefined'
? Symbol.for('@xen-orchestra/log')
: '@@@xen-orchestra/log'
global[symbol] = log => transport(log)
export const configure = config => {
transport = createTransport(config)
}
// -------------------------------------------------------------------
export const catchGlobalErrors = logger => {
// patch process
const onUncaughtException = error => {
logger.error('uncaught exception', { error })
}
const onUnhandledRejection = error => {
logger.warn('possibly unhandled rejection', { error })
}
const onWarning = error => {
logger.warn('Node warning', { error })
}
process.on('uncaughtException', onUncaughtException)
process.on('unhandledRejection', onUnhandledRejection)
process.on('warning', onWarning)
// patch EventEmitter
const EventEmitter = require('events')
const { prototype } = EventEmitter
const { emit } = prototype
function patchedEmit (event, error) {
event === 'error' && !this.listenerCount(event)
? logger.error('unhandled error event', { error })
: emit.apply(this, arguments)
}
prototype.emit = patchedEmit
return () => {
process.removeListener('uncaughtException', onUncaughtException)
process.removeListener('unhandledRejection', onUnhandledRejection)
process.removeListener('warning', onWarning)
if (prototype.emit === patchedEmit) {
prototype.emit = emit
}
}
}

View File

@ -0,0 +1,65 @@
import createTransport from './transports/console'
import LEVELS from './levels'
const symbol =
typeof Symbol !== 'undefined'
? Symbol.for('@xen-orchestra/log')
: '@@@xen-orchestra/log'
if (!(symbol in global)) {
// the default behavior, without requiring `configure` is to avoid
// logging anything unless it's a real error
const transport = createTransport()
global[symbol] = log => log.level > LEVELS.WARN && transport(log)
}
// -------------------------------------------------------------------
function Log (data, level, namespace, message, time) {
this.data = data
this.level = level
this.namespace = namespace
this.message = message
this.time = time
}
function Logger (namespace) {
this._namespace = namespace
// bind all logging methods
for (const name in LEVELS) {
const lowerCase = name.toLowerCase()
this[lowerCase] = this[lowerCase].bind(this)
}
}
const { prototype } = Logger
for (const name in LEVELS) {
const level = LEVELS[name]
prototype[name.toLowerCase()] = function (message, data) {
global[symbol](new Log(data, level, this._namespace, message, new Date()))
}
}
prototype.wrap = function (message, fn) {
const logger = this
const warnAndRethrow = error => {
logger.warn(message, { error })
throw error
}
return function () {
try {
const result = fn.apply(this, arguments)
const then = result != null && result.then
return typeof then === 'function'
? then.call(result, warnAndRethrow)
: result
} catch (error) {
warnAndRethrow(error)
}
}
}
const createLogger = namespace => new Logger(namespace)
export { createLogger }

View File

@ -0,0 +1,24 @@
const LEVELS = Object.create(null)
export { LEVELS as default }
// https://github.com/trentm/node-bunyan#levels
LEVELS.FATAL = 60 // service/app is going to down
LEVELS.ERROR = 50 // fatal for current action
LEVELS.WARN = 40 // something went wrong but it's not fatal
LEVELS.INFO = 30 // detail on unusual but normal operation
LEVELS.DEBUG = 20
export const NAMES = Object.create(null)
for (const name in LEVELS) {
NAMES[LEVELS[name]] = name
}
export const resolve = level => {
if (typeof level === 'string') {
level = LEVELS[level.toUpperCase()]
}
return level
}
Object.freeze(LEVELS)
Object.freeze(NAMES)

View File

@ -0,0 +1,32 @@
/* eslint-env jest */
import { forEach, isInteger } from 'lodash'
import LEVELS, { NAMES, resolve } from './levels'
describe('LEVELS', () => {
it('maps level names to their integer values', () => {
forEach(LEVELS, (value, name) => {
expect(isInteger(value)).toBe(true)
})
})
})
describe('NAMES', () => {
it('maps level values to their names', () => {
forEach(LEVELS, (value, name) => {
expect(NAMES[value]).toBe(name)
})
})
})
describe('resolve()', () => {
it('returns level values either from values or names', () => {
forEach(LEVELS, value => {
expect(resolve(value)).toBe(value)
})
forEach(NAMES, (name, value) => {
expect(resolve(name)).toBe(+value)
})
})
})

View File

@ -0,0 +1,20 @@
import LEVELS, { NAMES } from '../levels'
// Bind console methods (necessary for browsers)
const debugConsole = console.log.bind(console)
const infoConsole = console.info.bind(console)
const warnConsole = console.warn.bind(console)
const errorConsole = console.error.bind(console)
const { ERROR, INFO, WARN } = LEVELS
const consoleTransport = ({ data, level, namespace, message, time }) => {
const fn =
level < INFO
? debugConsole
: level < WARN ? infoConsole : level < ERROR ? warnConsole : errorConsole
fn('%s - %s - [%s] %s', time.toISOString(), namespace, NAMES[level], message)
data != null && fn(data)
}
export default () => consoleTransport

View File

@ -0,0 +1,68 @@
import prettyFormat from 'pretty-format' // eslint-disable-line node/no-extraneous-import
import { createTransport } from 'nodemailer' // eslint-disable-line node/no-extraneous-import
import { fromCallback } from 'promise-toolbox'
import { evalTemplate, required } from '../utils'
import { NAMES } from '../levels'
export default ({
// transport options (https://nodemailer.com/smtp/)
auth,
authMethod,
host,
ignoreTLS,
port,
proxy,
requireTLS,
secure,
service,
tls,
// message options (https://nodemailer.com/message/)
bcc,
cc,
from = required('from'),
to = required('to'),
subject = '[{{level}} - {{namespace}}] {{time}} {{message}}',
}) => {
const transporter = createTransport(
{
auth,
authMethod,
host,
ignoreTLS,
port,
proxy,
requireTLS,
secure,
service,
tls,
disableFileAccess: true,
disableUrlAccess: true,
},
{
bcc,
cc,
from,
to,
}
)
return log =>
fromCallback(cb =>
transporter.sendMail(
{
subject: evalTemplate(
subject,
key =>
key === 'level'
? NAMES[log.level]
: key === 'time' ? log.time.toISOString() : log[key]
),
text: prettyFormat(log.data),
},
cb
)
)
}

View File

@ -0,0 +1,42 @@
import splitHost from 'split-host' // eslint-disable-line node/no-extraneous-import node/no-missing-import
import { createClient, Facility, Severity, Transport } from 'syslog-client' // eslint-disable-line node/no-extraneous-import node/no-missing-import
import { fromCallback } from 'promise-toolbox'
import { startsWith } from 'lodash'
import LEVELS from '../levels'
// https://github.com/paulgrove/node-syslog-client#syslogseverity
const LEVEL_TO_SEVERITY = {
[LEVELS.FATAL]: Severity.Critical,
[LEVELS.ERROR]: Severity.Error,
[LEVELS.WARN]: Severity.Warning,
[LEVELS.INFO]: Severity.Informational,
[LEVELS.DEBUG]: Severity.Debug,
}
const facility = Facility.User
export default target => {
const opts = {}
if (target !== undefined) {
if (startsWith(target, 'tcp://')) {
target = target.slice(6)
opts.transport = Transport.Tcp
} else if (startsWith(target, 'udp://')) {
target = target.slice(6)
opts.transport = Transport.Ucp
}
;({ host: target, port: opts.port } = splitHost(target))
}
const client = createClient(target, opts)
return log =>
fromCallback(cb =>
client.log(log.message, {
facility,
severity: LEVEL_TO_SEVERITY[log.level],
})
)
}

View File

@ -0,0 +1,62 @@
import escapeRegExp from 'lodash/escapeRegExp'
// ===================================================================
const TPL_RE = /\{\{(.+?)\}\}/g
export const evalTemplate = (tpl, data) => {
const getData =
typeof data === 'function' ? (_, key) => data(key) : (_, key) => data[key]
return tpl.replace(TPL_RE, getData)
}
// -------------------------------------------------------------------
const compileGlobPatternFragment = pattern =>
pattern
.split('*')
.map(escapeRegExp)
.join('.*')
export const compileGlobPattern = pattern => {
const no = []
const yes = []
pattern.split(/[\s,]+/).forEach(pattern => {
if (pattern[0] === '-') {
no.push(pattern.slice(1))
} else {
yes.push(pattern)
}
})
const raw = ['^']
if (no.length !== 0) {
raw.push('(?!', no.map(compileGlobPatternFragment).join('|'), ')')
}
if (yes.length !== 0) {
raw.push('(?:', yes.map(compileGlobPatternFragment).join('|'), ')')
} else {
raw.push('.*')
}
raw.push('$')
return new RegExp(raw.join(''))
}
// -------------------------------------------------------------------
export const required = name => {
throw new Error(`missing required arg ${name}`)
}
// -------------------------------------------------------------------
export const serializeError = error => ({
...error,
message: error.message,
name: error.name,
stack: error.stack,
})

View File

@ -0,0 +1,13 @@
/* eslint-env jest */
import { compileGlobPattern } from './utils'
describe('compileGlobPattern()', () => {
it('works', () => {
const re = compileGlobPattern('foo, ba*, -bar')
expect(re.test('foo')).toBe(true)
expect(re.test('bar')).toBe(false)
expect(re.test('baz')).toBe(true)
expect(re.test('qux')).toBe(false)
})
})

View File

@ -0,0 +1 @@
dist/transports

View File

@ -1,31 +0,0 @@
#!/usr/bin/env node
'use strict'
// ===================================================================
// Better stack traces if possible.
require('../better-stacks')
// Use Bluebird for all promises as it provides better performance and
// less memory usage.
global.Promise = require('bluebird')
// Make unhandled rejected promises visible.
process.on('unhandledRejection', function (reason) {
console.warn('[Warn] Possibly unhandled rejection:', reason && reason.stack || reason)
})
;(function (EE) {
var proto = EE.prototype
var emit = proto.emit
proto.emit = function patchedError (event, error) {
if (event === 'error' && !this.listenerCount(event)) {
return console.warn('[Warn] Unhandled error event:', error && error.stack || error)
}
return emit.apply(this, arguments)
}
})(require('events').EventEmitter)
require('exec-promise')(require('../'))

View File

@ -1,11 +0,0 @@
'use strict'
// ===================================================================
// Enable xo logs by default.
if (process.env.DEBUG === undefined) {
process.env.DEBUG = 'app-conf,xo:*,-xo:api'
}
// Import the real main module.
module.exports = require('./dist').default

View File

@ -16,6 +16,9 @@
"url": "https://github.com/vatesfr/xen-orchestra.git"
},
"preferGlobal": true,
"bin": {
"xo-server": "dist/cli"
},
"files": [
"better-stacks.js",
"bin/",

176
packages/xo-server/src/cli.js Executable file
View File

@ -0,0 +1,176 @@
#!/usr/bin/env node
const APP_NAME = 'xo-server'
// Enable xo logs by default.
if (process.env.DEBUG === undefined) {
process.env.DEBUG = 'app-conf,xo:*,-xo:api'
}
// -------------------------------------------------------------------
require('@xen-orchestra/log/configure').configure([
{
filter: process.env.DEBUG,
level: 'warn',
transport: require('@xen-orchestra/log/transports/console').default(),
},
])
const { info, warn } = require('@xen-orchestra/log').createLogger('bootstrap')
process.on('unhandledRejection', reason => {
warn('possibly unhandled rejection', reason)
})
process.on('warning', warning => {
warn('Node warning', warning)
})
;(({ prototype }) => {
const { emit } = prototype
prototype.emit = function (event, error) {
event === 'error' && !this.listenerCount(event)
? warn('unhandled error event', error)
: emit.apply(this, arguments)
}
})(require('events').EventEmitter)
// Use Bluebird for all promises as it provides better performance and
// less memory usage.
const Bluebird = require('bluebird')
Bluebird.config({
longStackTraces: true,
warnings: true,
})
global.Promise = Bluebird
// -------------------------------------------------------------------
const main = async args => {
if (args.includes('--help') || args.includes('-h')) {
const { name, version } = require('../package.json')
return console.log(`Usage: ${name} [--safe-mode]
${name} v${version}`)
}
info('starting')
const config = await require('app-conf').load(APP_NAME, {
appDir: `${__dirname}/..`,
ignoreUnknownFormats: true,
})
// Print a message if deprecated entries are specified.
;['users', 'servers'].forEach(entry => {
if (entry in config) {
warn(`${entry} configuration is deprecated`)
}
})
const httpServer = require('stoppable')(new (require('http-server-plus'))())
const readFile = Bluebird.promisify(require('fs').readFile)
await Promise.all(
config.http.listen.map(
async ({
certificate,
// The properties was called `certificate` before.
cert = certificate,
key,
...opts
}) => {
if (cert !== undefined && key !== undefined) {
;[opts.cert, opts.key] = await Promise.all([
readFile(cert),
readFile(key),
])
}
try {
const niceAddress = await httpServer.listen(opts)
info(`web server listening on ${niceAddress}`)
} catch (error) {
if (error.niceAddress !== undefined) {
warn(`web server could not listen on ${error.niceAddress}`)
const { code } = error
if (code === 'EACCES') {
warn(' access denied.')
warn(' ports < 1024 are often reserved to privileges users.')
} else if (code === 'EADDRINUSE') {
warn(' address already in use.')
}
} else {
warn('web server could not listen', error)
}
}
}
)
)
// Now the web server is listening, drop privileges.
try {
const { group, user } = config
if (group !== undefined) {
process.setgid(group)
info('group changed to', group)
}
if (user !== undefined) {
process.setuid(user)
info('user changed to', user)
}
} catch (error) {
warn('failed to change group/user', error)
}
const child = require('child_process').fork(require.resolve('./worker.js'))
child.send([''])
const App = require('./xo').default
const app = new App({
appName: APP_NAME,
config,
httpServer,
safeMode: require('lodash/includes')(args, '--safe-mode'),
})
// Register web server close on XO stop.
app.on('stop', () => Bluebird.fromCallback(cb => httpServer.stop(cb)))
await app.start()
// Trigger a clean job.
await app.clean()
// Gracefully shutdown on signals.
//
// TODO: implements a timeout? (or maybe it is the services launcher
// responsibility?)
require('lodash/forEach')(['SIGINT', 'SIGTERM'], signal => {
let alreadyCalled = false
process.on(signal, () => {
if (alreadyCalled) {
warn('forced exit')
process.exit(1)
}
alreadyCalled = true
info(`${signal} caught, closing…`)
app.stop()
})
})
await require('event-to-promise')(app, 'stopped')
}
main(process.argv.slice(2)).then(
() => info('bye :-)'),
error => {
if (typeof error === 'number') {
process.exitCode = error
} else {
warn('fatal error', error)
}
}
)

View File

@ -0,0 +1,348 @@
const compilePug = require('pug').compile
const createProxyServer = require('http-proxy').createServer
const JsonRpcPeer = require('json-rpc-peer')
const LocalStrategy = require('passport-local').Strategy
const parseCookies = require('cookie').parse
const Passport = require('passport')
const serveStatic = require('serve-static')
const WebSocket = require('ws')
const { fromCallback } = require('promise-toolbox')
const { invalidCredentials } = require('xo-common/api-errors')
const { readFile } = require('fs')
const proxyConsole = require('../proxy-console')
const { debug, warn } = require('@xen-orchestra/log').createLogger('front')
function createExpressApp ({ http: config }, httpServer) {
const express = require('express')()
express.use(require('helmet')())
if (config.redirectToHttps) {
const https = config.listen.find(
_ =>
_.port !== undefined &&
(_.cert !== undefined || _.certificate !== undefined)
)
if (https === undefined) {
warn('could not setup HTTPs redirection: no HTTPs config found')
} else {
const { port } = https
express.use((req, res, next) => {
if (req.secure) {
return next()
}
res.redirect(`https://${req.hostname}:${port}${req.originalUrl}`)
})
}
}
Object.keys(config.mounts).forEach(url => {
const paths = config.mounts[url]
;(Array.isArray(paths) ? paths : [paths]).forEach(path => {
debug('Setting up %s → %s', url, path)
express.use(url, serveStatic(path))
})
})
return express
}
function setUpApi (config, httpServer, xo) {
const webSocketServer = new WebSocket.Server({
noServer: true,
})
xo.on('stop', () => fromCallback(cb => webSocketServer.close(cb)))
const onConnection = (socket, upgradeReq) => {
const { remoteAddress } = upgradeReq.socket
debug('+ WebSocket connection (%s)', remoteAddress)
// Create the abstract XO object for this connection.
const connection = xo.createUserConnection()
connection.once('close', () => {
socket.close()
})
// Create the JSON-RPC server for this connection.
const jsonRpc = new JsonRpcPeer(message => {
if (message.type === 'request') {
return xo.callApiMethod(connection, message.method, message.params)
}
})
connection.notify = jsonRpc.notify.bind(jsonRpc)
// Close the XO connection with this WebSocket.
socket.once('close', () => {
debug('- WebSocket connection (%s)', remoteAddress)
connection.close()
})
// Connect the WebSocket to the JSON-RPC server.
socket.on('message', message => {
jsonRpc.write(message)
})
const onSend = error => {
if (error) {
warn('WebSocket send:', error.stack)
}
}
jsonRpc.on('data', data => {
// The socket may have been closed during the API method
// execution.
if (socket.readyState === WebSocket.OPEN) {
socket.send(data, onSend)
}
})
}
httpServer.on('upgrade', (req, socket, head) => {
if (req.url === '/api/') {
webSocketServer.handleUpgrade(req, socket, head, ws =>
onConnection(ws, req)
)
}
})
}
function setUpConsoleProxy (httpServer, xo) {
const webSocketServer = new WebSocket.Server({
noServer: true,
})
const CONSOLE_PROXY_PATH_RE = /^\/api\/consoles\/(.*)$/
httpServer.on('upgrade', async (req, socket, head) => {
const matches = CONSOLE_PROXY_PATH_RE.exec(req.url)
if (!matches) {
return
}
const [, id] = matches
try {
// TODO: factorize permissions checking in an Express middleware.
{
const { token } = parseCookies(req.headers.cookie)
const user = await xo.authenticateUser({ token })
if (!await xo.hasPermissions(user.id, [[id, 'operate']])) {
throw invalidCredentials()
}
const { remoteAddress } = socket
debug('+ Console proxy (%s - %s)', user.name, remoteAddress)
socket.on('close', () => {
debug('- Console proxy (%s - %s)', user.name, remoteAddress)
})
}
const xapi = xo.getXapi(id, ['VM', 'VM-controller'])
const vmConsole = xapi.getVmConsole(id)
// FIXME: lost connection due to VM restart is not detected.
webSocketServer.handleUpgrade(req, socket, head, connection => {
proxyConsole(connection, vmConsole, xapi.sessionId)
})
} catch (error) {
console.error((error && error.stack) || error)
}
})
}
async function setUpPassport (express, xo) {
// necessary for connect-flash
express.use(require('cookie-parser')())
express.use(
require('express-session')({
resave: false,
saveUninitialized: false,
// TODO: should be in the config file.
secret: 'CLWguhRZAZIXZcbrMzHCYmefxgweItKnS',
})
)
// necessary for Passport to display error messages
express.use(require('connect-flash')())
// necessary for Passport to access the username and password from the sign
// in form
express.use(require('body-parser').urlencoded({ extended: false }))
express.use(Passport.initialize())
const strategies = { __proto__: null }
xo.registerPassportStrategy = strategy => {
Passport.use(strategy)
const { name } = strategy
if (name !== 'local') {
strategies[name] = strategy.label || name
}
}
// Registers the sign in form.
const signInPage = compilePug(
await fromCallback(cb => readFile(`${__dirname}/../signin.pug`, cb))
)
express.get('/signin', (req, res, next) => {
res.send(
signInPage({
error: req.flash('error')[0],
strategies,
})
)
})
express.get('/signout', (req, res) => {
res.clearCookie('token')
res.redirect('/')
})
const SIGNIN_STRATEGY_RE = /^\/signin\/([^/]+)(\/callback)?(:?\?.*)?$/
express.use(async (req, res, next) => {
const { url } = req
const matches = url.match(SIGNIN_STRATEGY_RE)
if (matches !== null) {
return Passport.authenticate(matches[1], async (err, user, info) => {
if (err) {
return next(err)
}
if (!user) {
req.flash('error', info ? info.message : 'Invalid credentials')
return res.redirect('/signin')
}
// The cookie will be set in via the next request because some
// browsers do not save cookies on redirect.
req.flash(
'token',
(await xo.createAuthenticationToken({ userId: user.id })).id
)
// The session is only persistent for internal provider and if 'Remember me' box is checked
req.flash(
'session-is-persistent',
matches[1] === 'local' && req.body['remember-me'] === 'on'
)
res.redirect(req.flash('return-url')[0] || '/')
})(req, res, next)
}
const token = req.flash('token')[0]
if (token) {
const isPersistent = req.flash('session-is-persistent')[0]
if (isPersistent) {
// Persistent cookie ? => 1 year
res.cookie('token', token, { maxAge: 1000 * 60 * 60 * 24 * 365 })
} else {
// Non-persistent : external provider as Github, Twitter...
res.cookie('token', token)
}
next()
} else if (req.cookies.token) {
next()
} else if (
/favicon|fontawesome|images|styles|\.(?:css|jpg|png)$/.test(url)
) {
next()
} else {
req.flash('return-url', url)
return res.redirect('/signin')
}
})
// Install the local strategy.
xo.registerPassportStrategy(
new LocalStrategy(async (username, password, done) => {
try {
const user = await xo.authenticateUser({ username, password })
done(null, user)
} catch (error) {
done(null, false, { message: error.message })
}
})
)
}
function setUpProxies ({ http: { proxies } }, httpServer, express, xo) {
if (proxies === undefined) {
return
}
const proxy = createProxyServer({
ignorePath: true,
}).on('error', error => console.error(error))
const prefixes = Object.keys(proxies).sort((a, b) => a.length - b.length)
const n = prefixes.length
// HTTP request proxy.
express.use((req, res, next) => {
const { url } = req
for (let i = 0; i < n; ++i) {
const prefix = prefixes[i]
if (url.startsWith(prefix)) {
const target = proxies[prefix]
proxy.web(req, res, {
target: target + url.slice(prefix.length),
})
return
}
}
next()
})
// WebSocket proxy.
const webSocketServer = new WebSocket.Server({
noServer: true,
})
xo.on('stop', () => fromCallback(cb => webSocketServer.close(cb)))
httpServer.on('upgrade', (req, socket, head) => {
const { url } = req
for (let i = 0; i < n; ++i) {
const prefix = prefixes[i]
if (url.startsWith(prefix)) {
const target = proxies[prefix]
proxy.ws(req, socket, head, {
target: target + url.slice(prefix.length),
})
return
}
}
})
}
export default async function main ({ config, httpServer, safeMode }) {
const express = createExpressApp(config, httpServer)
setUpProxies(config, httpServer, express, xo)
setUpApi(config, httpServer, xo)
// must be set up before the API
setUpConsoleProxy(httpServer, xo)
await setUpPassport(express, xo)
// TODO: express.use(xo._handleHttpRequest.bind(xo))
}

View File

@ -1,656 +0,0 @@
import appConf from 'app-conf'
import bind from 'lodash/bind'
import blocked from 'blocked'
import createExpress from 'express'
import createLogger from 'debug'
import has from 'lodash/has'
import helmet from 'helmet'
import includes from 'lodash/includes'
import proxyConsole from './proxy-console'
import serveStatic from 'serve-static'
import startsWith from 'lodash/startsWith'
import stoppable from 'stoppable'
import WebSocket from 'ws'
import { compile as compilePug } from 'pug'
import { createServer as createProxyServer } from 'http-proxy'
import { fromEvent } from 'promise-toolbox'
import { join as joinPath } from 'path'
import JsonRpcPeer from 'json-rpc-peer'
import { invalidCredentials } from 'xo-common/api-errors'
import { ensureDir, readdir, readFile } from 'fs-extra'
import WebServer from 'http-server-plus'
import Xo from './xo'
import {
forEach,
isArray,
isFunction,
mapToArray,
pFromCallback,
} from './utils'
import bodyParser from 'body-parser'
import connectFlash from 'connect-flash'
import cookieParser from 'cookie-parser'
import expressSession from 'express-session'
import passport from 'passport'
import { parse as parseCookies } from 'cookie'
import { Strategy as LocalStrategy } from 'passport-local'
// ===================================================================
const debug = createLogger('xo:main')
const warn = (...args) => {
console.warn('[Warn]', ...args)
}
// ===================================================================
const DEPRECATED_ENTRIES = ['users', 'servers']
async function loadConfiguration () {
const config = await appConf.load('xo-server', {
appDir: joinPath(__dirname, '..'),
ignoreUnknownFormats: true,
})
debug('Configuration loaded.')
// Print a message if deprecated entries are specified.
forEach(DEPRECATED_ENTRIES, entry => {
if (has(config, entry)) {
warn(`${entry} configuration is deprecated.`)
}
})
return config
}
// ===================================================================
function createExpressApp () {
const app = createExpress()
app.use(helmet())
// Registers the cookie-parser and express-session middlewares,
// necessary for connect-flash.
app.use(cookieParser())
app.use(
expressSession({
resave: false,
saveUninitialized: false,
// TODO: should be in the config file.
secret: 'CLWguhRZAZIXZcbrMzHCYmefxgweItKnS',
})
)
// Registers the connect-flash middleware, necessary for Passport to
// display error messages.
app.use(connectFlash())
// Registers the body-parser middleware, necessary for Passport to
// access the username and password from the sign in form.
app.use(bodyParser.urlencoded({ extended: false }))
// Registers Passport's middlewares.
app.use(passport.initialize())
return app
}
async function setUpPassport (express, xo) {
const strategies = { __proto__: null }
xo.registerPassportStrategy = strategy => {
passport.use(strategy)
const { name } = strategy
if (name !== 'local') {
strategies[name] = strategy.label || name
}
}
// Registers the sign in form.
const signInPage = compilePug(
await readFile(joinPath(__dirname, '..', 'signin.pug'))
)
express.get('/signin', (req, res, next) => {
res.send(
signInPage({
error: req.flash('error')[0],
strategies,
})
)
})
express.get('/signout', (req, res) => {
res.clearCookie('token')
res.redirect('/')
})
const SIGNIN_STRATEGY_RE = /^\/signin\/([^/]+)(\/callback)?(:?\?.*)?$/
express.use(async (req, res, next) => {
const { url } = req
const matches = url.match(SIGNIN_STRATEGY_RE)
if (matches) {
return passport.authenticate(matches[1], async (err, user, info) => {
if (err) {
return next(err)
}
if (!user) {
req.flash('error', info ? info.message : 'Invalid credentials')
return res.redirect('/signin')
}
// The cookie will be set in via the next request because some
// browsers do not save cookies on redirect.
req.flash(
'token',
(await xo.createAuthenticationToken({ userId: user.id })).id
)
// The session is only persistent for internal provider and if 'Remember me' box is checked
req.flash(
'session-is-persistent',
matches[1] === 'local' && req.body['remember-me'] === 'on'
)
res.redirect(req.flash('return-url')[0] || '/')
})(req, res, next)
}
const token = req.flash('token')[0]
if (token) {
const isPersistent = req.flash('session-is-persistent')[0]
if (isPersistent) {
// Persistent cookie ? => 1 year
res.cookie('token', token, { maxAge: 1000 * 60 * 60 * 24 * 365 })
} else {
// Non-persistent : external provider as Github, Twitter...
res.cookie('token', token)
}
next()
} else if (req.cookies.token) {
next()
} else if (
/favicon|fontawesome|images|styles|\.(?:css|jpg|png)$/.test(url)
) {
next()
} else {
req.flash('return-url', url)
return res.redirect('/signin')
}
})
// Install the local strategy.
xo.registerPassportStrategy(
new LocalStrategy(async (username, password, done) => {
try {
const user = await xo.authenticateUser({ username, password })
done(null, user)
} catch (error) {
done(null, false, { message: error.message })
}
})
)
}
// ===================================================================
async function registerPlugin (pluginPath, pluginName) {
const plugin = require(pluginPath)
const { description, version = 'unknown' } = (() => {
try {
return require(pluginPath + '/package.json')
} catch (_) {
return {}
}
})()
// Supports both “normal” CommonJS and Babel's ES2015 modules.
const {
default: factory = plugin,
configurationSchema,
configurationPresets,
testSchema,
} = plugin
// The default export can be either a factory or directly a plugin
// instance.
const instance = isFunction(factory)
? factory({
xo: this,
getDataDir: () => {
const dir = `${this._config.datadir}/${pluginName}`
return ensureDir(dir).then(() => dir)
},
})
: factory
await this.registerPlugin(
pluginName,
instance,
configurationSchema,
configurationPresets,
description,
testSchema,
version
)
}
const debugPlugin = createLogger('xo:plugin')
function registerPluginWrapper (pluginPath, pluginName) {
debugPlugin('register %s', pluginName)
return registerPlugin.call(this, pluginPath, pluginName).then(
() => {
debugPlugin(`successfully register ${pluginName}`)
},
error => {
debugPlugin(`failed register ${pluginName}`)
debugPlugin(error)
}
)
}
const PLUGIN_PREFIX = 'xo-server-'
const PLUGIN_PREFIX_LENGTH = PLUGIN_PREFIX.length
async function registerPluginsInPath (path) {
const files = await readdir(path).catch(error => {
if (error.code === 'ENOENT') {
return []
}
throw error
})
await Promise.all(
mapToArray(files, name => {
if (startsWith(name, PLUGIN_PREFIX)) {
return registerPluginWrapper.call(
this,
`${path}/${name}`,
name.slice(PLUGIN_PREFIX_LENGTH)
)
}
})
)
}
async function registerPlugins (xo) {
await Promise.all(
mapToArray(
[`${__dirname}/../node_modules/`, '/usr/local/lib/node_modules/'],
xo::registerPluginsInPath
)
)
}
// ===================================================================
async function makeWebServerListen (
webServer,
{
certificate,
// The properties was called `certificate` before.
cert = certificate,
key,
...opts
}
) {
if (cert && key) {
;[opts.cert, opts.key] = await Promise.all([readFile(cert), readFile(key)])
}
try {
const niceAddress = await webServer.listen(opts)
debug(`Web server listening on ${niceAddress}`)
} catch (error) {
if (error.niceAddress) {
warn(`Web server could not listen on ${error.niceAddress}`)
const { code } = error
if (code === 'EACCES') {
warn(' Access denied.')
warn(' Ports < 1024 are often reserved to privileges users.')
} else if (code === 'EADDRINUSE') {
warn(' Address already in use.')
}
} else {
warn('Web server could not listen:', error.message)
}
}
}
async function createWebServer ({ listen, listenOptions }) {
const webServer = stoppable(new WebServer())
await Promise.all(
mapToArray(listen, opts =>
makeWebServerListen(webServer, { ...listenOptions, ...opts })
)
)
return webServer
}
// ===================================================================
const setUpProxies = (express, opts, xo) => {
if (!opts) {
return
}
const proxy = createProxyServer({
ignorePath: true,
}).on('error', error => console.error(error))
// TODO: sort proxies by descending prefix length.
// HTTP request proxy.
express.use((req, res, next) => {
const { url } = req
for (const prefix in opts) {
if (startsWith(url, prefix)) {
const target = opts[prefix]
proxy.web(req, res, {
target: target + url.slice(prefix.length),
})
return
}
}
next()
})
// WebSocket proxy.
const webSocketServer = new WebSocket.Server({
noServer: true,
})
xo.on('stop', () => pFromCallback(cb => webSocketServer.close(cb)))
express.on('upgrade', (req, socket, head) => {
const { url } = req
for (const prefix in opts) {
if (startsWith(url, prefix)) {
const target = opts[prefix]
proxy.ws(req, socket, head, {
target: target + url.slice(prefix.length),
})
return
}
}
})
}
// ===================================================================
const setUpStaticFiles = (express, opts) => {
forEach(opts, (paths, url) => {
if (!isArray(paths)) {
paths = [paths]
}
forEach(paths, path => {
debug('Setting up %s → %s', url, path)
express.use(url, serveStatic(path))
})
})
}
// ===================================================================
const setUpApi = (webServer, xo, verboseLogsOnErrors) => {
const webSocketServer = new WebSocket.Server({
noServer: true,
})
xo.on('stop', () => pFromCallback(cb => webSocketServer.close(cb)))
const onConnection = (socket, upgradeReq) => {
const { remoteAddress } = upgradeReq.socket
debug('+ WebSocket connection (%s)', remoteAddress)
// Create the abstract XO object for this connection.
const connection = xo.createUserConnection()
connection.once('close', () => {
socket.close()
})
// Create the JSON-RPC server for this connection.
const jsonRpc = new JsonRpcPeer(message => {
if (message.type === 'request') {
return xo.callApiMethod(connection, message.method, message.params)
}
})
connection.notify = bind(jsonRpc.notify, jsonRpc)
// Close the XO connection with this WebSocket.
socket.once('close', () => {
debug('- WebSocket connection (%s)', remoteAddress)
connection.close()
})
// Connect the WebSocket to the JSON-RPC server.
socket.on('message', message => {
jsonRpc.write(message)
})
const onSend = error => {
if (error) {
warn('WebSocket send:', error.stack)
}
}
jsonRpc.on('data', data => {
// The socket may have been closed during the API method
// execution.
if (socket.readyState === WebSocket.OPEN) {
socket.send(data, onSend)
}
})
}
webServer.on('upgrade', (req, socket, head) => {
if (req.url === '/api/') {
webSocketServer.handleUpgrade(req, socket, head, ws =>
onConnection(ws, req)
)
}
})
}
// ===================================================================
const CONSOLE_PROXY_PATH_RE = /^\/api\/consoles\/(.*)$/
const setUpConsoleProxy = (webServer, xo) => {
const webSocketServer = new WebSocket.Server({
noServer: true,
})
xo.on('stop', () => pFromCallback(cb => webSocketServer.close(cb)))
webServer.on('upgrade', async (req, socket, head) => {
const matches = CONSOLE_PROXY_PATH_RE.exec(req.url)
if (!matches) {
return
}
const [, id] = matches
try {
// TODO: factorize permissions checking in an Express middleware.
{
const { token } = parseCookies(req.headers.cookie)
const user = await xo.authenticateUser({ token })
if (!await xo.hasPermissions(user.id, [[id, 'operate']])) {
throw invalidCredentials()
}
const { remoteAddress } = socket
debug('+ Console proxy (%s - %s)', user.name, remoteAddress)
socket.on('close', () => {
debug('- Console proxy (%s - %s)', user.name, remoteAddress)
})
}
const xapi = xo.getXapi(id, ['VM', 'VM-controller'])
const vmConsole = xapi.getVmConsole(id)
// FIXME: lost connection due to VM restart is not detected.
webSocketServer.handleUpgrade(req, socket, head, connection => {
proxyConsole(connection, vmConsole, xapi.sessionId)
})
} catch (error) {
console.error((error && error.stack) || error)
}
})
}
// ===================================================================
const USAGE = (({ name, version }) => `Usage: ${name} [--safe-mode]
${name} v${version}`)(require('../package.json'))
// ===================================================================
export default async function main (args) {
if (includes(args, '--help') || includes(args, '-h')) {
return USAGE
}
{
const debug = createLogger('xo:perf')
blocked(
ms => {
debug('blocked for %sms', ms | 0)
},
{
threshold: 50,
}
)
}
const config = await loadConfiguration()
const webServer = await createWebServer(config.http)
// Now the web server is listening, drop privileges.
try {
const { user, group } = config
if (group) {
process.setgid(group)
debug('Group changed to', group)
}
if (user) {
process.setuid(user)
debug('User changed to', user)
}
} catch (error) {
warn('Failed to change user/group:', error)
}
// Creates main object.
const xo = new Xo(config)
// Register web server close on XO stop.
xo.on('stop', () => pFromCallback(cb => webServer.stop(cb)))
// Connects to all registered servers.
await xo.start()
// Trigger a clean job.
await xo.clean()
// Express is used to manage non WebSocket connections.
const express = createExpressApp()
if (config.http.redirectToHttps) {
let port
forEach(config.http.listen, listen => {
if (listen.port && (listen.cert || listen.certificate)) {
port = listen.port
return false
}
})
if (port === undefined) {
warn('Could not setup HTTPs redirection: no HTTPs port found')
} else {
express.use((req, res, next) => {
if (req.secure) {
return next()
}
res.redirect(`https://${req.hostname}:${port}${req.originalUrl}`)
})
}
}
// Must be set up before the API.
setUpConsoleProxy(webServer, xo)
// Must be set up before the API.
express.use(bind(xo._handleHttpRequest, xo))
// Everything above is not protected by the sign in, allowing xo-cli
// to work properly.
await setUpPassport(express, xo)
// Attaches express to the web server.
webServer.on('request', express)
webServer.on('upgrade', (req, socket, head) => {
express.emit('upgrade', req, socket, head)
})
// Must be set up before the static files.
setUpApi(webServer, xo, config.verboseApiLogsOnErrors)
setUpProxies(express, config.http.proxies, xo)
setUpStaticFiles(express, config.http.mounts)
if (!includes(args, '--safe-mode')) {
await registerPlugins(xo)
}
// Gracefully shutdown on signals.
//
// TODO: implements a timeout? (or maybe it is the services launcher
// responsibility?)
forEach(['SIGINT', 'SIGTERM'], signal => {
let alreadyCalled = false
process.on(signal, () => {
if (alreadyCalled) {
warn('forced exit')
process.exit(1)
}
alreadyCalled = true
debug('%s caught, closing…', signal)
xo.stop()
})
})
await fromEvent(xo, 'stopped')
debug('bye :-)')
}

View File

@ -0,0 +1,3 @@
process.on('message', ([action, ...args]) => {
console.log(action, args)
})

View File

@ -0,0 +1,143 @@
import blocked from 'blocked'
import { createLogger } from '@xen-orchestra/log'
import { fromEvent } from 'promise-toolbox'
import { ensureDir, readdir } from 'fs-extra'
import Xo from './xo'
// ===================================================================
const { debug } = createLogger('xo:main')
// ===================================================================
async function registerPlugin (pluginPath, pluginName) {
const plugin = require(pluginPath)
const { description, version = 'unknown' } = (() => {
try {
return require(pluginPath + '/package.json')
} catch (_) {
return {}
}
})()
// Supports both “normal” CommonJS and Babel's ES2015 modules.
const {
default: factory = plugin,
configurationSchema,
configurationPresets,
testSchema,
} = plugin
// The default export can be either a factory or directly a plugin
// instance.
const instance =
typeof factory === 'function'
? factory({
xo: this,
getDataDir: () => {
const dir = `${this._config.datadir}/${pluginName}`
return ensureDir(dir).then(() => dir)
},
})
: factory
await this.registerPlugin(
pluginName,
instance,
configurationSchema,
configurationPresets,
description,
testSchema,
version
)
}
const debugPlugin = createLogger('xo:plugin')
function registerPluginWrapper (pluginPath, pluginName) {
debugPlugin('register %s', pluginName)
return registerPlugin.call(this, pluginPath, pluginName).then(
() => {
debugPlugin(`successfully register ${pluginName}`)
},
error => {
debugPlugin(`failed register ${pluginName}`)
debugPlugin(error)
}
)
}
const PLUGIN_PREFIX = 'xo-server-'
const PLUGIN_PREFIX_LENGTH = PLUGIN_PREFIX.length
async function registerPluginsInPath (path) {
const files = await readdir(path).catch(error => {
if (error.code === 'ENOENT') {
return []
}
throw error
})
await Promise.all(
files.map(name => {
if (name.startsWith(PLUGIN_PREFIX)) {
return registerPluginWrapper.call(
this,
`${path}/${name}`,
name.slice(PLUGIN_PREFIX_LENGTH)
)
}
})
)
}
async function registerPlugins (xo) {
await Promise.all(
[`${__dirname}/../node_modules/`, '/usr/local/lib/node_modules/'].map(
xo::registerPluginsInPath
)
)
}
// ===================================================================
async function main ({ config, safeMode }) {
{
const debug = createLogger('xo:perf')
blocked(ms => {
debug('blocked for %sms', ms | 0)
})
}
// Creates main object.
const xo = new Xo(config)
// Connects to all registered servers.
await xo.start()
// Trigger a clean job.
await xo.clean()
if (!safeMode) {
await registerPlugins(xo)
}
await new Promise(resolve => {
const onMessage = message => {
if (message[0] === 'STOP') {
process.removeListener('message', onMessage)
resolve()
}
}
process.on('message', onMessage)
})
await fromEvent(xo, 'stopped')
}
main().then(
() => process.send(['STOPPED']),
error => process.send(['STOPPED_WITH_ERROR', error])
)