compute_id : undefined _create: (cb) => _open_somewhere: (ssh_timeout, cb) => # assuming that the project is *not* "zpool imported'" on any host, choose a host and set everything up there. # Willi create the project if that hasn't happened yet. dbg = (m) -> @dbg("_open_somewhere",[ssh_timeout],m) cb() _close_all_but_one: (open_state, cb) => # Assuming that the project currently is "zpool imported" on *multiple* hosts, choose the one with most # recent activity, and export the others (also sending their streams, and saving to the db.... just in case). dbg = (m) -> @dbg("_close_all_but_one",open_state,m) cb() _verify_project_works_at: (compute_id, ssh_timeout, cb) => # cb(err) # Determine whether or not it is possible to ssh from this host to the compute host # and login as the project user within ssh_timeout seconds. dbg = (m) -> @dbg("_verify_project_works_at",[compute_id, ssh_timeout],m) cb() open: (opts) => opts = defaults opts, ssh_timeout : 15 cb : required # cb(err, ip_address of computer on which project is ready to use) dbg = (m) -> @dbg("open_project",[],m) state = undefined compute_id = undefined works = undefined host = undefined async.series([ (cb) => dbg('get state') @state (err, _state) => state = _state cb(err) (cb) => open = (x for x in state when x.import_pool?) dbg('already open on #{open.length} hosts') if open.length == 0 @_open_somewhere opts.ssh_timeout, (err, _compute_id) => if err cb(err) else compute_id = _compute_id works = true cb() else if open.length == 1 compute_id = open[0].compute_id cb() else @_close_all_but_one open, (err, _compute_id) => compute_id = _compute_id cb(err) (cb) => if works cb() else dbg('verify that project really works at @{compute_id}') @_verify_project_works_at compute_id, opts.ssh_timeout, (err) => if err dbg("it doesn't work at @{compute_id}") @_open_somewhere (err, _compute_id) => compute_id = _compute_id cb(err) else dbg("yep, it works") cb() (cb) => compute_id_to_host compute_id, (err, _host) => host = _host; cb(err) ], (err) => if not err @compute_id = compute_id opts.cb(err, host) ) ############# ### # pool must be an array [['hostname', port, 'token_key...'], ...] exports.client = (opts) -> # cb(err, client) opts = defaults opts, pool : required cb : required C = new Client pool, (err) -> if err cb(err) else cb(false, C) class Client constructor : (@pool, cb) -> # cb(err) @connections = [] f = (i, cb) -> h = @hosts[i] @_connect(h[0], h[1], h[2], cb) async.map @pool, f, (err) -> cb(err) _connect: (hostname, port, token_key, cb) => socket = misc_node.connect_to_locked_socket host : @hosts[i][0] port : @hosts[i][1] timeout : 15 token : @hosts[i][2] cb : (err) -> if err cb(err) else misc_node.enable_mesg(socket) @connections.push(socket) cb() snap_ls : (opts) => snap_restore : (opts) => snap_log : (opts) => ### ### if req.method == "POST" data = '' req.on 'data', (chunk) -> data += chunk req.on 'end', () -> winston.debug("data = ", data) winston.debug("got file of length #{data.length} and id #{query.uuid}, #{misc.to_json()}") ### ocreate_unix_user_on_random_compute_server = (opts) -> opts = defaults opts, cb : required opts.cb(false, {host:'localhost', username:'sage0', port:22}) """ TO DO async.series([ (cb) -> # first get a computer on which to create an account # (TODO: load balance) database.random_compute_server cb : (err, _compute_server) -> if err cb(err) else compute_server = _compute_server cb() ], (err) -> cb(err, unix_user) ) """ # The CodeMirrorDiffSyncHub class represents a global hub viewed as a # remote server for this client. class CodeMirrorDiffSyncHub constructor: (@cm_session) -> connect: (remote) => @remote = remote recv_edits: (edit_stack, last_version_ack, cb) => #console.log("Sending the following edits to the server: ", edit_stack, last_version_ack) @cm_session.call message : message.codemirror_diffsync(edit_stack:edit_stack, last_version_ack:last_version_ack) timeout : 5 cb : (err, mesg) => #console.log("codemirror sync got back: ", mesg) if err cb(err) else if mesg.event == 'error' cb(mesg.error) else @remote.recv_edits(mesg.edit_stack, mesg.last_version_ack, cb) class CodeMirrorSession extends EventEmitter constructor : (opts) -> opts = defaults opts, conn : required project_id : required path : required cb : required @conn = opts.conn @project_id = opts.project_id @path = opts.path @connect (err, session, content) => if not err @init_dsync(content) opts.cb(err, session, content) connect: (cb) => @conn.call message: message.codemirror_get_session path : @path project_id : @project_id cb : (err, resp) => if err cb(err) else if resp.event == 'error' cb(resp.error) else @session_uuid = resp.session_uuid cb(false, @, resp.content) init_dsync: (content) => @dsync_client = new diffsync.DiffSync(doc:content) @dsync_server = new CodeMirrorDiffSyncHub(@) @dsync_client.connect(@dsync_server) @dsync_server.connect(@dsync_client) @conn.on 'codemirror_diffsync_ready', (mesg) => if mesg.session_uuid == @session_uuid before = @dsync_client.live @sync (err) => if before != @dsync_client.live @emit 'change', @dsync_client.live @sync() sync: (cb) => if @_syncing # TODO -- ensure that this lock times out -- we do not want to loose sync forever!! cb("already syncing") return @_syncing = true @dsync_client.push_edits (err) => @_syncing = false cb?(err) change: (content, cb) => if @dsync_client? @dsync_client.live = content @sync (err) => if err cb?(err) else cb?(false, @dsync_client.live) else cb?() call: (opts) => opts = defaults opts, message : required timeout : 10 cb : undefined #console.log("call #{misc.to_json(opts.message)}") opts.message.session_uuid = @session_uuid @conn.call message : opts.message timeout : opts.timeout cb : (err, result) => if not err and result.event == 'reconnect' # Try one time to connect then resend message. @connect (err) => if err opts.cb(err) else @conn.call(message: opts.message, cb:opts.cb) return opts.cb(err, result) write_to_disk: (cb) => @call message: message.codemirror_write_to_disk() cb : cb read_from_disk: (cb) => @call message : message.codemirror_read_from_disk() cb : cb get_content: (cb) => # cb(err, content) @call message : message.codemirror_get_content() cb : (err, resp) => cb(err, resp?.content) ########################################################### # Receive and process the edits from the other end of the sync connection. recv_edits: (edit_stack, last_version_ack, cb) => if SIMULATE_LOSS and Math.random() < .5 # Simulate packet loss console.log("Simulating loss!"); cb(true); return # Keep only edits that we still need to send. @edit_stack = (edits for edits in @edit_stack when edits.shadow_version > last_version_ack) if edit_stack.length == 0 cb() return if edit_stack[0].shadow_version != @shadow_version and edit_stack[0].shadow_version == @backup_shadow_version # Lost return packet @shadow = @_copy(@backup_shadow) @shadow_version = @backup_shadow_version @edit_stack = [] # Make a backup, just in case it turns out that our message # back to the client that we applied these changes is lost # "Lost return packet." @backup_shadow = @_copy(@shadow) @backup_shadow_version = @shadow_version tasks = [] for edits in edit_stack tasks.push( (cb) => if edits.shadow_version == @shadow_version if edits.shadow_checksum != @_checksum(@shadow) # Data corruption in memory or network: we have to just restart everything from scratch. cb('reset') return @last_version_received = edits.shadow_version @shadow = @_apply_edits0(edits.edits, @shadow) @shadow_version += 1 @live = @_apply_edits0(edits.edits, @live) cb() # @_apply_edits edits.edits, @shadow, (err, result) => # if err # cb(err) # else # @shadow = result # @shadow_version += 1 # @_apply_edits_to_live(edits.edits, cb) else if edits.shadow_version < @shadow_version # Packet duplication or loss: ignore -- it will sort itself out later. cb() else if edits.shadow_version > @shadow_version # This should be impossible, unless there is data corruption. cb('reset') ) async.series(tasks, (err) => cb?(err)) # Connect the client with a codemirror file editing session. connect_to_codemirror_session: (opts) => opts = defaults opts, client : required project_id : required params : undefined session_uuid : undefined # if undefined, a new session is created; if defined, connect to session or get error cb : required # cb(err, [session_connected message]) if not opts.session_uuid? opts.session_uuid = uuid.v4() @_open_session_socket session_uuid : opts.session_uuid project_id : opts.project_id type : 'codemirror' params : opts.params cb : (err, socket) => winston.debug("open codemirror session -- #{err}") if err opts.cb(err) return if not socket.clients? socket.clients = [] else socket.clients.push(opts.client) socket.on 'end', () => delete @_sockets[opts.session_uuid] socket.on 'mesg', (type, mesg) => winston.debug("GOT A MESSAGE FROM THE CodeMirror local-hub server: #{misc.to_json(mesg)}") mesg = message.session_connected(session_uuid : opts.session_uuid) opts.cb(false, mesg) ############################################################################ ############################################################################ class old_Project constructor: (@project_id) -> if not @project_id? throw "When creating Project, the project_id must be defined" @local_path = "#{project_data}/#{@project_id}/" owner: (cb) => database.get_project_data project_id : @project_id columns : ['account_id'] cb : (err, result) => if err cb(err) else cb(err, result[0]) ############################################## # Database state stuff ############################################## _choose_new_host: (cb) -> # For now we just choose a host at random. In the long run, # we may experiment with other load balancing algorithms, and # take into account properties of the project itself. For # example, we will have an "infinite uptime" option. database.random_compute_server type:'project', cb:(err, hostinfo) -> #winston.debug("*** #{err}, #{misc.to_json(hostinfo)}") if err cb(err) else if not hostinfo? cb("No project servers are currently available.") else cb(false, hostinfo.host) _minus_one_host: (host, cb) -> database.score_compute_server(host:host, cb:cb, delta:-1) _plus_one_host: (host, cb) -> database.score_compute_server(host:host, cb:cb, delta:+1) _connect: (host, cb) -> if not host? throw "BUG -- host must be defined" if @_socket? and @_socket.host == host cb(false, @_socket.socket) return socket = net.connect {host:host, port:cass.COMPUTE_SERVER_PORTS.project}, () => winston.debug("!! connected to #{misc.to_json(host)}") # connected misc_node.enable_mesg(socket) # We cache the connection for later. We only cache a # connection to *one* host, since a project is only # supposed to ever be hosted on one compute server. @_socket = {host:host, socket:socket} cb(false, socket) # There are numerous actions that require adding a listener to # the socket, sending blobs, then removing it. Since many of # these could happen simultaneously, I'm upping the max # listeners a bit, just in case. socket.setMaxListeners(30) socket.once 'error', (err) => if err == 'ECONNREFUSED' # An error occured connecting -- this suggests # that the relevant project server is down. cb(err) # Make sure not to cache the socket in case anything goes wrong with it. socket.once 'close', () => if @_socket? delete @_socket socket.once 'end', () => if @_socket? delete @_socket socket.once 'timeout', () => if @_socket? delete @_socket # Returns the host that is currently hosting the project # (according to the database) or null if the project is not # currently on any host. get_host: (cb) => # cb(err, host) database.get_project_host(project_id:@project_id, cb:cb) _is_being_opened: (cb) => # cb(err, is_being_opened) database.is_project_being_opened(project_id:@project_id, cb:cb) _lock_for_opening: (ttl, cb) => # cb(err) database.lock_project_for_opening(project_id:@project_id, ttl:ttl, cb:cb) _remove_opening_lock: (cb) => database.remove_project_opening_lock(project_id:@project_id, cb:cb) _is_being_saved: (cb) => # cb(err, is_being_opened) database.is_project_being_saved(project_id:@project_id, cb:cb) _lock_for_saving: (ttl, cb) => # cb(err) database.lock_project_for_saving(project_id:@project_id, ttl:ttl, cb:cb) _remove_saving_lock: (cb) => database.remove_project_saving_lock(project_id:@project_id, cb:cb) # Open the project (if necessary) and get a working socket connection to the project_server. socket: (cb) -> # cb(err, socket) host = undefined socket = undefined async.series([ (c) => @open (err, _host) => if err winston.debug("project.socket -- error opening project -- #{to_json(err)}") c(err) else host = _host winston.debug("project.socket -- deploy to host '#{host}'") c() (c) => @_connect host, (err, _socket) => if err winston.debug("project.socket -- error connecting to project server #{err}") c(err) else socket = _socket winston.debug("project.socket -- got a socket connection to the project_server") c() ], (err) -> if err cb(err) else cb(false, socket) ) # Open the project on some host if it is not already opened. If it # is currently being opened by this or another hub, we return an # error. (The client can choose to show some sort of notification # and retry after waiting a moment.) open: (cb) -> # cb(err, host) if not cb? throw "cb must be defined" host = undefined had_to_recurse = false async.series([ # First, we check in the database to see if the project is # already opened on a compute server, and if so, return # that host. (c) => winston.debug("open project -- get host") @get_host (err, _host) -> host = _host if err c(true) # failed -- done else if _host? c(true) # done else c() # Is anybody else trying to open this project right now? (c) => winston.debug("open project -- check if anybody else opening project right now") @_is_being_opened (err, is_being_opened) => if err c(true) # failed -- don't try any further else if is_being_opened # In the rare case of two opens at the same # time, try again repeatedly, up to a total of # 10 seconds (after which user client will # have given up). total_time = 0 delta = 200 try_again = () => @get_host (err, _host) -> if not err and _host? # got it! host = _host c(true) else delta += delta if total_time + delta <= 10000 setTimeout(try_again, delta) else # give up. c("Project #{@project_id} is currently being opened by another hub. Please try again in a few seconds. (#{err})") setTimeout(try_again, delta) else # Not open, not being opened, let's get on it! # The 15 below is a timeout in seconds, after which our lock self-destructs. @_lock_for_opening(15, c) # We choose a project server host. (c) => winston.debug("open project -- choose a host") @_choose_new_host (err, _host) => if err @_remove_opening_lock() c(err) else host = _host c() # Open the project on that host (c) => winston.debug("open project -- open on that host (='#{host}')") @_open_on_host host, (err) => @_remove_opening_lock() if err if host == 'localhost' or host == '127.0.0.1' # debugging mode -- just give up instantly. winston.debug("_open_on_host -- err opening on localhost '#{to_json(err)}'") host = undefined # so error messages will propagate below. c(err) return # Downvote this host, and try again. @_minus_one_host host, (err) => if err # This is serious -- if we can't even connect to the database # to flag a host as down, then there is no point in going on. host = undefined # so error messages will propagate below. c(true) else # Try again. This will not lead to infinite recursion since each # time it is called, we just successfully flagged a host as down # in the database, and eventually we'll run out of hosts. had_to_recurse = true @open(cb) host = undefined # so error messages will propagate below. c() else # Finally, got it! # Upvote this host, since it worked. @_plus_one_host(host) c() ], (err) -> if not had_to_recurse # the recursive call will call cb if host? # if host got defined then done cb(false, host) else cb(err) ) _initialize_local_path: (cb) => # Initialize a local copy of the project -- first the simplest possible approach; this will get more complicated later ! TODO async.series([ (cb) => fs.mkdir @local_path, 0o700, cb (cb) => # Just do a local rsync from the template. misc_node.execute_code command : "rsync" args : ['-axH', PROJECT_TEMPLATE, @local_path] timeout : 30 bash : false path : SALVUS_HOME cb : cb (cb) => # Add a gitconfig, if not in local files gitconfig = @local_path + '.gitconfig' fs.exists gitconfig, (exists) => if exists cb() return @owner((err,account_id) => if err cb(err) else database.get_gitconfig account_id: account_id, cb:(err, contents) => if err cb(err) else fs.writeFile(gitconfig, contents, cb) ) ], cb) _open_on_host: (host, cb) => # Find out if project already exists on some remote storage server # If yes, rsync it from there to local # If no, create project from scratch (locally) # or create project from a template (another project; like "forking" in # github) by rsyncing from remote storage server # In either case, we now have local project -- select a project_server. # Rsync local project to that project_server. remote_path = "#{misc_node.username(@project_id)}@#{host}:" id = uuid.v4() # used to tag communication with the project server socket = undefined quota = undefined idle_timeout = undefined async.series([ (cb) => winston.debug("project: initialize '#{@local_path}'") fs.exists @local_path, (exists) => if not exists @_initialize_local_path(cb) else cb() # Get a connection to the project server. (cb) => winston.debug("_open_on_host - get a connection to the project server.") @_connect host, (err, s) -> if err cb(err) else socket = s cb() # Get meta information about the project that is needed to open the project. (cb) => winston.debug("_open_on_host -- get meta information about the project that is needed to open the project.") database.get_project_open_info project_id:@project_id, cb:(err, result) -> if err cb(err) else {quota, idle_timeout} = result cb() # Send open_project mesg. (cb) => winston.debug("_open_on_host -- Send open_project mesg") # TODO: ssh_public_key = "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCd5N5Vz1Kv4QqYkWVGxKQfvnZjlMnbdu2wkB9mInLqUAoqkkOTq4B0Qq5rtgMK8AfTYqoi4jAF1R/qJpGLGqAm359Yc6vXOMq5Gy0cDekAb/JOADbmtsYzElv+YhCwnD1MAXqjrv0+W16Y8j6IFjjuvzf4Egnria0s5kddcCwLCkYU9sJUs6OXUDqzK05jhO3PF3P40f3sKKFdZLIH4rVkKdM3ynJ0gNoBqI/xak/9Tz3U0aSZXK7DfO7kjI5PCZr2EVI83u8XAwskDM9XDdifNED3Ctae/x7FkpdpLNiXjOsUKJ0cnJmBFDwO0M4Vt8L8zROy2aH8ZEhD1cqGF69D salvus" mesg_open_project = message.open_project id : id project_id : @project_id quota : quota idle_timeout : idle_timeout ssh_public_key: ssh_public_key # computed on startup of hub socket.write_mesg 'json', mesg_open_project cb() # Wait for response that project user has been created and public key written (cb) => winston.debug("_open_on_host -- wait for the project server to respond") socket.recv_mesg id:id, type:'json', timeout:30, cb:(mesg) -> winston.debug("_open_on_host -- received response #{misc.to_json(mesg)}") switch mesg.event when 'error' # something went wrong... cb(mesg.error) when 'project_opened' # finally, got it. cb() else cb("Expected a 'project_opened' message, but got a '#{mesg.event}' message instead.") # Push files to the project (cb) => winston.debug("_open_on_host -- push files to the project") misc_node.execute_code command : "rsync" args : ['-axH', '--delete', '--exclude', '.ssh', '-e', "ssh -o StrictHostKeyChecking=no", @local_path, remote_path] timeout : 30 bash : false path : SALVUS_HOME cb : (err, output) => if err cb("Error running rsync at all: #{err}") else cb() # Save where project is running to the database (cb) => winston.debug("_open_on_host -- save where (='#{host}') project running to database") database.set_project_host(project_id:@project_id, host:host, cb:cb) ], cb) # This is called by 'open' once we determine that the project is # not already opened, and also we determine on which host we plan # to open the project. XXX_open_on_host: (host, cb) -> socket = undefined id = uuid.v4() # used to tag communication with the project server bundles = undefined quota = undefined idle_timeout = undefined async.series([ # Get a connection to the project server. (c) => winston.debug("_open_on_host - get a connection to the project server.") @_connect host, (err, s) -> if err c(err) else socket = s c() # Get each bundle blob from the database in preparation to # send it to the project_server. (c) => winston.debug("_open_on_host - get bundle blobs from the database.") database.get_project_bundles project_id:@project_id, cb:(err, result) -> if err c(err) else # bundles is an array of pairs [filename, Buffer]. bundles = result # Make a corresponding list of temporary # uuid's that will be used when sending the # bundles. bundles.uuids = ([uuid.v4(),bundles[i][0]] for i in [0...bundles.length]) c() # Get meta information about the project that is needed to open the project. (c) => winston.debug("_open_on_host -- get meta information about the project that is needed to open the project.") database.get_project_open_info project_id:@project_id, cb:(err, result) -> if err c(err) else {quota, idle_timeout} = result c() # Send open_project mesg. (c) => winston.debug("_open_on_host -- Send open_project mesg") mesg_open_project = message.open_project id : id project_id : @project_id bundle_uuids : bundles.uuids quota : quota idle_timeout : idle_timeout socket.write_mesg 'json', mesg_open_project c() # Starting sending the bundles as blobs. (c) -> winston.debug("_open_on_host -- start sending #{bundles.length} bundles as blobs") for i in [0...bundles.length] socket.write_mesg 'blob', {uuid:bundles.uuids[i][0], blob:bundles[i][1]} c() # Wait for the project server to respond with success # (having received all blobs) or failure (something went wrong). (c) -> winston.debug("_open_on_host -- wait for the project server to respond") socket.recv_mesg id:id, type:'json', timeout:30, cb:(mesg) -> winston.debug("_open_on_host -- received response #{misc.to_json(mesg)}") switch mesg.event when 'error' # something went wrong... c(mesg.error) when 'project_opened' # finally, got it. c() else c("Expected a 'project_opened' message, but got a '#{mesg.event}' message instead.") # Save where project is running to the database (c) => winston.debug("_open_on_host -- save where (='#{host}') project running to database") database.set_project_host(project_id:@project_id, host:host, cb:c) ], cb) # Save the project to the database. This involves saving at least # zero (!) bundles to the project_bundles uuid:blob table. save: (cb) -> # cb(err) -- indicates when done id = uuid.v4() # used to tag communication with the project server host = undefined not_open = false nothing_to_do = false async.series([ # If project is already locked for saving (by this or # another hub), return an error. (c) => @_is_being_saved (err, is_being_saved) => if err c(err) else if is_being_saved nothing_to_do = true c(true) else @_lock_for_saving(15, c) # Determine which project_server is hosting this project. # If none, then there is nothing further to do. (c) => database.get_project_host project_id:@project_id, cb:(err, _host) => if err c(err) else if not _host? not_open = true c(true) else host = _host c() (c) => winston.debug("save: rsync files from remote") remote_path = "#{misc_node.username(@project_id)}@#{host}:" misc_node.execute_code command : "rsync" args : ['-axH', '--stats', '--delete', '--exclude', '.ssh', '-e', "ssh -o StrictHostKeyChecking=no", remote_path, @local_path] timeout : 30 bash : false path : SALVUS_HOME cb : (err, output) => if err c("rsync project --> hub: exited with nonzero code: stderr = #{err}") else s = 'Number of files transferred:' i = output.stdout.indexOf(s) if i == -1 winston.debug("BUG! Output of rsync doesn't contain '#{s}', but it must!") else t = output.stdout.slice(i+s.length+1) j = t.indexOf('\n') if t.slice(0,j) != '0' # Sent at least one file modification. # Get the size of the resulting directory, and save this to the database. @size_of_local_copy((err,size) => if err c(err) else database.touch_project project_id : @project_id size : size cb : c ) else c() ], (err) => if err and nothing_to_do cb?() return @_remove_saving_lock() if not_open cb?() else cb?(err) ) # return total size in bytes of the local copy of files in the project size_of_local_copy: (cb) => misc_node.execute_code command : 'du' args : ['-bs', @local_path] timeout : 15 bash : false path : SALVUS_HOME cb : (err, output) => if err cb(err) else i = output.stdout.indexOf(' ') size = parseInt(output.stdout.slice(0,i)) cb(false, size) delete: (cb) => @close(() => database.delete_project(project_id:@project_id, cb:cb)) # Close the project. This save the project first, then # kills all processes and clears all disk space. close: (cb) => # cb(err) -- indicates when done id = uuid.v4() socket = undefined host = undefined async.series([ (c) => @save(c) # Get project's current host (c) => database.get_project_host project_id:@project_id, cb: (err, _host) -> if err c(true) else if not _host? # not currently hosted somewhere, so "done" but no error. c('ok') else host = _host c() # Connect to the project server that is hosting this project right now. (c) => @_connect host, (err, s) -> if err c(err) else socket = s c() # Put a lock, so nobody tries to save or open this project while we are closing. # 15 seconds should be plenty of time. (c) => winston.debug("project close: -- lock") @_lock_for_opening(15, ->) @_lock_for_saving(15, c) # Send the close message (c) => winston.debug("project close: -- send close message") socket.write_mesg 'json', message.close_project(id: id, project_id: @project_id) c() # And wait for response from project server (c) => winston.debug("project close: -- waiting for response from project server") socket.recv_mesg type:'json', id:id, timeout:10, cb:(mesg) => switch mesg.event when 'error' c(mesg.error) when 'project_closed' c() # successfully closed project else c("BUG closing project -- unknown mesg event type '#{mesg.event}'") # Store in the database that the project is not allocated # on any host. (c) => winston.debug("project close -- store that project is closed") database.set_project_host project_id : @project_id host : "" cb : c ], (err) => @_remove_opening_lock() @_remove_saving_lock() if err == 'ok' cb() else cb(err) ) # Tag the input message mesg with a uuid and this project_id, then # send it to the project server hosting this project (or create # one if there is none right now). Finally do the standard: # cb(err, response_mesg). If err is true, then there was an error # getting a socket to the project server in the first place; if # err is false, then it is still possible that response_mesg.event # == 'error' due to the project server having issues. call: (opts) => opts = defaults opts, message : required timeout : 10 cb : required socket = undefined id = uuid.v4() async.series([ (c) => @socket (err,_socket) -> if err opts.cb(err) c(true) else socket = _socket c() (c) => if not opts.message.id? opts.message.id = id opts.message.project_id = @project_id socket.write_mesg 'json', opts.message c() (c) => socket.recv_mesg type:'json', id:opts.message.id, timeout:opts.timeout, cb:(mesg) -> opts.cb(false, mesg) c() ]) # Read a file from a project into memory on the hub. This is # used, e.g., for client-side editing, worksheets, etc. This does # not pull the file from the database; instead, it loads it live # from the project_server virtual machine. read_file: (opts) -> # cb(err, content_of_file) -- indicates when done {path, archive, cb} = defaults opts, path : required archive : undefined cb : required socket = undefined id = uuid.v4() data = undefined data_uuid = undefined async.series([ # Get a socket connection to the project host. This will open # the project if it isn't already opened. (c) => @socket (err, _socket) -> if err c(err) else socket = _socket c() (c) => socket.write_mesg 'json', message.read_file_from_project(id:id, project_id:@project_id, path:path, archive:archive) socket.recv_mesg type:'json', id:id, timeout:10, cb:(mesg) => switch mesg.event when 'error' c(mesg.error) when 'file_read_from_project' data_uuid = mesg.data_uuid c() else c("Unknown mesg event '#{mesg.event}'") (c) => socket.recv_mesg type: 'blob', id:data_uuid, timeout:10, cb:(_data) -> data = _data c() ], (err) -> if err cb(err) else cb(false, data) ) # Write a file to a compute node. This is used when saving during # client-side editing, for worksheets, etc. This does not # directly change anything in the database -- it only impacts the # files on the compute node. This does not trigger a save (which # would change the database). write_file: (path, data, cb) -> # cb(err) socket = undefined id = uuid.v4() data_uuid = uuid.v4() async.series([ (c) => @socket (err, _socket) -> #winston.debug("@socket returned: #{err}, #{_socket}") if err c(err) else socket = _socket c() (c) => mesg = message.write_file_to_project id : id project_id : @project_id path : path data_uuid : data_uuid # winston.debug("mesg = #{misc.to_json(mesg)}") socket.write_mesg 'json', mesg socket.write_mesg 'blob', {uuid:data_uuid, blob:data} c() (c) => socket.recv_mesg type: 'json', id:id, timeout:10, cb:(mesg) -> switch mesg.event when 'file_written_to_project' c() when 'error' c(mesg.error) else c("Unexpected message type '#{mesg.event}'") ], cb) make_directory: (path, cb) -> socket = undefined id = uuid.v4() async.series([ (c) => @socket (err, _socket) -> if err c(err) else socket = _socket (c) => m = message.make_directory_in_project id : id project_id : @project_id path : path socket.write_mesg 'json', m c() (c) => socket.recv_mesg type:'json', id:id, timeout:10, cb:(mesg) -> switch mesg.event when 'directory_made_in_project' c() when 'error' c(mesg.error) else c("Unexpected message type '#{mesg.event}'") ], cb) # Move a file or directory move_file: (src, dest, cb) -> winston.debug("MOVE: #{src} --> #{dest}") socket = undefined id = uuid.v4() async.series([ (c) => @socket (err, _socket) -> if err c(err) else socket = _socket c() (c) => m = message.move_file_in_project id : id project_id : @project_id src : src dest : dest socket.write_mesg 'json', m c() (c) => socket.recv_mesg type:'json', id:id, timeout:10, cb:(mesg) -> switch mesg.event when 'file_moved_in_project' c() when 'error' c(mesg.error) else c("Unexpected message type '#{mesg.event}'") ], cb) # Remove a file or directory remove_file: (path, cb) => id = uuid.v4() @call message: message.remove_file_from_project id : id project_id : @project_id path : path cb : cb # Branch op branch_op : (branch, op, cb) => @call message: message["#{op}_project_branch"] branch : branch project_id : @project_id cb : cb # Exec a command exec : (mesg, cb) => @call message : mesg cb : cb timeout : mesg.timeout ######################################## git_add_all: (opts) => opts = defaults opts, project_id : required cb : undefined @exec project_id : opts.project_id command : 'git' args : ['add', '--all'] cb : opts.cb git_commit_all: (opts) => # Commit all modified files that are under revision control to the repo. opts = defaults opts, project_id : required author : required message : "Saved all files." cb : undefined nothing_to_do = false async.series([ (cb) => # Check to see if there are uncommited changes @exec project_id : opts.project_id command : 'git' args : ['status'] cb : (err, output) -> if err cb(err) else if output.exit_code cb(output.stdout + output.stderr) else if output.stdout.indexOf('nothing to commit') != -1 # DONE -- nothing further to do nothing_to_do = true cb(true) else # Add and commit as usual. cb() (cb) => @exec project_id : opts.project_id command : 'git' args : ['commit', '-a', '-m', opts.message, '--author', opts.author] cb : (err, output) -> if err cb(err) else if output.exit_code cb(err + " -- " + misc.to_json(output)) else cb() ], (err) => if err and not nothing_to_do opts.cb("Error commiting all changed files under control to the repository -- #{err}") else opts.cb() # good ) ######## # Commit all changes (cb) -> winston.debug("save_project -- commit") commit user : username(mesg.project_id) path : path commit_mesg : mesg.commit_mesg gitconfig : mesg.gitconfig add_all : mesg.add_all cb : cb @cassandra.select table : @_table columns : ['value'] where : {name:@opts.name, uuid:opts.uuid} cb : (err, results) => if err opts.cb(err) else if results.length == 0 opts.cb("no object with uuid #{opts.uuid}") else console.log(results[0][0]) @cassandra.update table : @_table where : {name:@opts.name, uuid:opts.uuid} set : {value:@_to_db(@_from_db(results[0][0]))} ttl : opts.ttl cb : opts.cb #@cql("UPDATE compute_servers SET score = score + ? WHERE host = ?", [opts.delta, opts.host], opts.cb) if mesg.path[mesg.path.length-1] == '/' errmesg = message.error id : mesg.id error : 'path must be a filename, not a directory name (it must not end in "/")' socket.write_mesg('json', errmesg) return < # Choose a random unused unix UID, where unused means by definition # that /home/UID does not exist. We then create that directory and # call cb with it. There is once in a billion years possibility of a # race condition, so we do not worry about it. Our uid is a random # number between 1100 and 2^32-2=4294967294. create_user = (cb) -> uid = misc.randint(1100, 4294967294) fs.exists("/home/#{uid}", (exists) -> if exists create_user(cb) # This recursion won't blow stack, since # it is highly unlikely to ever happen. else cmd = "useradd -U -m #{project_uuid}" ) # fs.mkdir(, , () -> # fs.chown( #` child_process.exec("useradd -U -m #{project_uuid}", (error, stdout, stderr) ->cb()) uid = misc.randint(1100, 4294967294) fs.exists("/home/#{uid}", (exists) -> if exists choose_uid(cb) # This recursion won't blow stack, since # it is highly unlikely to ever happen. else cmd = "useradd -u #{uid} -g #{uid} -m -U #{project_uuid}" path = "/home/#{project_uuid}" fs.mkdir(, , () -> fs.chown( -------------------------- # buf = null # console_session.on('data', (data) -> # if buf == null # buf = data # else # buf = Buffer.concat([buf, data]) # if buf[buf.length-1].toString() = '\u0000' # client.push_to_client(from_json(buf[buf.length-1].toString())) # inform client of successful connection ################################################# # HTML parsing functionality -- probably move ################################################# # require('async') htmlparser = require("htmlparser") # extract plain text from a dom tree object, as produced by htmlparser. dom_to_text = (dom, divs=false) -> result = '' for d in dom switch d.type when 'text' result += d.data when 'tag' switch d.name when 'div' divs = true result += '\n' when 'br' if not divs result += '\n' if d.children? result += dom_to_text(d.children, divs) return result # create a lossy plain text representation of html exports.html_to_text = (opts) -> opts = defaults opts, html : required cb : required # cb(error, result) handler = new htmlparser.DefaultHandler (error, dom) -> if error opts.cb(error) else opts.cb(false, dom_to_text(dom)) (new htmlparser.Parser(handler)).parseComplete(opts.html) ####################################################### # Accounts, plans, auth, account events, etc. ####################################################### class Plans # EXAMPLES: # c = new (require("cassandra").Cassandra)(['localhost']); plans = c.plans() # d = plans.create_plan('demo') # d.data = "a demo plan" # d.save() # d.number_of_accounts_with_this_plan(console.log) # plans.newest_plan('demo', console.log) constructor: (@conn) -> create_plan: (name) -> p = @plan(uuid.v4()) p.name = name p.save() @conn.cql("UPDATE newest_plans SET plan_id=? WHERE name=?", [p.plan_id, name]) return p plan: (plan_id) -> return new Plan(@conn, plan_id) newest_plan: (name, cb) -> # cb(the newest plan with given name | undefined) @conn.cql("SELECT plan_id FROM newest_plans WHERE name=?", [name], (e,r) => cb(if r.length>0 then @plan(r[0].get('plan_id').value))) class Plan constructor: (@conn, @plan_id, @data) -> save: () -> @conn.cql("UPDATE plans SET name=?, data=?, time=? WHERE plan_id=?", [@name, JSON.stringify(@data), @time, @plan_id], (e,r) -> console.log(e); console.log(r)) number_of_accounts_with_this_plan: (cb) -> @conn.cql("SELECT COUNT(*) FROM accounts WHERE plan_id=?", [@plan_id], (e,r) -> cb(r[0].get('count').value)) class Accounts constructor: (@conn) -> class Account constructor: (@conn, @account_id) -> class Auth constructor: (@conn, @account_id, @provider, @login_name, @info) -> save: () -> @conn.cql("UPDATE auths SET info=? WHERE account_id=? AND provider=? AND login_name=?", [@info, @account_id, @provider, @login_name]) --- (err, results) -> cb(if results.length == 1 then from_json(results[0].get('value').value))) (err, results) -> cb(if results.length == 1 then from_json(results[0].get('value').value))) q = "" vals = [] for key, val of props q += "#{key}=?," vals.push(val) q = q.slice(0,-1) vals.push(account_id) @conn.cql("UPDATE accounts SET #{q} WHERE account_id=?", vals, cb) --- # # TODO TODO TODO ### # TESTING: see if cql connection works. tm = (new Date()).getTime() cassandra.cql("SELECT * FROM sage_servers", [], (err, results) -> res.end("#{(new Date()).getTime() - tm}\n#{err} #{results}")) ### ### sage_conn = new sage.Connection( host:'localhost' port:10000 recv:(mesg) -> winston.log("sage: received message #{mesg}") cb: -> sage_conn.send(message.start_session()) sage_conn.send(message.execute_code(0,"factor(2012)")) ) ### class TincConf(object): """ Generate and store all the tinc configuration files needed by a node in a private temp directory, which is deleted when this instance goes out of scope. Use obj.files() to get a mapping filename:absolute_path_to_file. """ def __init__(self, ip_address): path = tempfile.mkdtemp() self._path = path open(os.path.join(path, 'tinc-up'),'w').write("#!/bin/sh\nifconfig $INTERFACE %s netmask 255.255.0.0"%ip_address) open(os.path.join(path, 'tinc.conf'),'w').write("Name = %s\nConnectTo = %s"%(ip_address, HOSTNAME)) sh['tincd', '--config', path, '-K'] open(os.path.join(path, ip_address),'w').write( "Subnet = %s/32\n%s"%(ip_address,open(os.path.join(path, 'rsa_key.priv')).read().strip())) self._files = dict([(file, os.path.join(path, file)) for file in ['tinc-up', 'tinc.conf', ip_address, 'rsa_key.pub']]) def files(self): return self._files def __del__(self): shutil.rmtree(self._path) ---------------------- elif line.startswith('<'): # use hosts listed in first column of non-commented lines in the named file for x in open(os.path.join(os.path.split(filename)[0], line.strip(' <>'))).readlines(): l = x.split('#')[0].strip().split() if l: groups[group].append(l[0]) -------------------- class CassandraConf(object): def __init__(self, filename): self._conf = [] for r in open(filename).readlines(): line = r.split('#')[0].strip() # ignore comments and leading/trailing whitespace if line: hostname, location, token = line.split() self._conf.append({'hostname':hostname, 'location':location, 'token':int(token)}) def nodes(self): return [x['hostname'] for x in self._conf] ------------- ########################################### # memcache usage ########################################### import memcache, tornadoasyncmemcache MEMCACHE_SERVERS = ["127.0.0.1:11211"] cache = memcache.Client(MEMCACHE_SERVERS) class MemCache(object): """ Use memcache to implement a simple key:value store. * Keys are hashed, but verified for correctness on read; hash collision implies only 1 object with that hash is stored. * Setting is done asynchronously (in that data is sent async), but reading is done synchronously by default, though get_async is also supported. """ def __init__(self, namespace): self._cache = memcache.Client(MEMCACHE_SERVERS) self._async_cache = tornadoasyncmemcache.ClientPool(MEMCACHE_SERVERS, maxclients=256) self._ram_cache = {} self._namespace = namespace def key(self, key): return self._namespace + '.' + str(hash(key)) def get_async(self, key, callback=None): # async get: result=self[key], then call the callback with result or # call callback with None if there is no such key. def f(result): if result is not None and result[0] == key: callback(result[1]) else: callback(None) self._async_cache.get(self.key(key), callback=f) def __getitem__(self, key): # sync get: self[key] c = self._cache.get(self.key(key)) if c is not None and c[0] == key: return c[1] def __setitem__(self, key, result): # async set: self[key] = result key = key.strip() self._async_cache.set(self.key(key), (key, result), callback=lambda data:None) stateless_execution_cache = MemCache('stateless') ####################################################################### # the below is subject to change a lot and none is implemented. # client --> hub # All files are added, directory is commited, new bundle is created, bundle is saved to database. message event : 'save_project' id : required session_id : required # hub --> client message event : 'project_saved' id : required session_id : required # client --> hub message event : 'create_sage_session' id : required project_id : required # hub --> client message event : 'sage_session_created' id : required project_id : required session_id : required # client --> hub --> sage_session message event : 'message_sage_session' # send message to a Sage session id : required session_id : required message : required # the message for the Sage session # client --> hub message event : 'get_directory_tree' id : required project_id : required # hub --> client message event : 'directory_tree' id : required project_id : required tree : required # object that gives the directory tree of files in this project # client --> hub message event : 'git_command' id : required project_id : required command : required ################################################# # HTML parsing functionality -- probably move ################################################# htmlparser = require("htmlparser") # extract plain text from a dom tree object, as produced by htmlparser. dom_to_text = (dom, divs=false) -> result = '' for d in dom switch d.type when 'text' result += d.data when 'tag' switch d.name when 'div' divs = true result += '\n' when 'br' if not divs result += '\n' if d.children? result += dom_to_text(d.children, divs) return result # create a lossy plain text representation of html exports.html_to_text = (opts) -> opts = defaults opts, html : required cb : required # cb(error, result) handler = new htmlparser.DefaultHandler (error, dom) -> if error opts.cb(error) else opts.cb(false, dom_to_text(dom)) (new htmlparser.Parser(handler)).parseComplete(opts.html)