From 04b92945fe66322014bd5657b712d53d7d818e30 Mon Sep 17 00:00:00 2001 From: Zaid Abdulla Date: Sat, 12 Jan 2013 16:08:15 +0300 Subject: [PATCH] Renamed profiles to presets. Added preconditition. --- acs.coffee | 24 ++++----- common.coffee | 19 +++++++ config.coffee | 4 +- db.coffee | 8 +-- mongo-query.coffee | 80 +++++++++++++++++++++++++++++ presets.coffee | 115 ++++++++++++++++++++++++++++++++++++++++++ profiles.coffee | 123 --------------------------------------------- 7 files changed, 230 insertions(+), 143 deletions(-) create mode 100644 mongo-query.coffee create mode 100644 presets.coffee delete mode 100644 profiles.coffee diff --git a/acs.coffee b/acs.coffee index 719f672..28a56b4 100644 --- a/acs.coffee +++ b/acs.coffee @@ -6,7 +6,7 @@ tr069 = require './tr-069' tasks = require './tasks' sanitize = require('./sanitize').sanitize db = require './db' -profiles = require './profiles' +presets = require './presets' currentRequest = null @@ -14,7 +14,7 @@ currentRequest = null applyConfigurations = (taskList) -> if taskList.length - util.log("#{currentRequest.deviceId}: Profiles discrepancy found") + util.log("#{currentRequest.deviceId}: Presets discrepancy found") db.tasksCollection.save(taskList, (err) -> for task in taskList util.log("#{currentRequest.deviceId}: Added #{task.name} task #{task._id}") @@ -120,19 +120,19 @@ nextTask = (cookies) -> resParams = {} if not task - # no more taks, check profile discrepancy - db.memcached.get(["#{currentRequest.deviceId}_profiles_hash", 'profiles_hash'], (err, results) -> - profilesHash = results['profiles_hash'] - deviceProfilesHash = results["#{currentRequest.deviceId}_profiles_hash"] + # no more taks, check presets discrepancy + db.memcached.get(["#{currentRequest.deviceId}_presets_hash", 'presets_hash'], (err, results) -> + presetsHash = results['presets_hash'] + devicePresetsHash = results["#{currentRequest.deviceId}_presets_hash"] - if not deviceProfilesHash or profilesHash != deviceProfilesHash - profiles.findDeviceConfigurationDiscrepancy(currentRequest.deviceId, profilesHash, (taskList) -> + if not devicePresetsHash or presetsHash != devicePresetsHash + presets.assertPresets(currentRequest.deviceId, presetsHash, (taskList) -> applyConfigurations(taskList) ) - else if not profilesHash - profiles.getProfilesHash((hash) -> - if hash != deviceProfilesHash - profiles.findDeviceConfigurationDiscrepancy(currentRequest.deviceId, profilesHash, (taskList) -> + else if not presetsHash + presets.getPresetsHash((hash) -> + if hash != devicePresetsHash + presets.assertPresets(currentRequest.deviceId, presetsHash, (taskList) -> applyConfigurations(taskList) ) else diff --git a/common.coffee b/common.coffee index 2ad4b22..4ddb123 100644 --- a/common.coffee +++ b/common.coffee @@ -1,3 +1,14 @@ +exports.UNDEFINED_TYPE = '[object Undefined]' +exports.NULL_TYPE = '[object Null]' +exports.NUMBER_TYPE = '[object Number]' +exports.STRING_TYPE = '[object String]' +exports.ARRAY_TYPE = '[object Array]' +exports.OBJECT_TYPE = '[object Object]' + +exports.typeOf = (obj) -> + Object.prototype.toString.call(obj) + + exports.endsWith = (str, suffix) -> str.indexOf(suffix, str.length - suffix.length) isnt -1 @@ -16,7 +27,15 @@ exports.getDeviceId = (deviceIdStruct) -> return "#{escape(deviceIdStruct['OUI'])}-#{escape(deviceIdStruct['SerialNumber'])}" + exports.extend = (obj, mixin) -> obj[name] = method for name, method of mixin obj + +exports.getParamValueFromPath = (obj, path) -> + pp = path.split('.') + ref = obj + for p in pp + ref = ref[p] + return ref diff --git a/config.coffee b/config.coffee index a017f7e..e44bb88 100644 --- a/config.coffee +++ b/config.coffee @@ -4,8 +4,8 @@ exports.ACS_INTERFACE = '0.0.0.0' exports.API_PORT = 7557 exports.API_INTERFACE = '0.0.0.0' exports.CACHE_DURATION = 60 # in seconds -exports.PROFILES_CACHE_DURATION = 86400 -exports.PROFILES_TIME_PADDING = 1 +exports.PRESETS_CACHE_DURATION = 86400 +exports.PRESETS_TIME_PADDING = 1 exports.MONGODB_SOCKET = '/tmp/mongodb-27017.sock' exports.MEMCACHED_SOCKET = '/tmp/memcached.sock' exports.WORKER_RESPAWN_TIME = 3000 diff --git a/db.coffee b/db.coffee index 89bc1e6..88c147b 100644 --- a/db.coffee +++ b/db.coffee @@ -4,10 +4,6 @@ Memcached = require 'memcached' Memcached.config.maxExpiration = 86400 memcached = new Memcached(config.MEMCACHED_SOCKET) -# Create MongoDB connections -tasksCollection = null -devicesCollection = null -profilesCollection = null dbserver = new mongo.Server(config.MONGODB_SOCKET, 0, {auto_reconnect: true}) db = new mongo.Db(config.DATABASE_NAME, dbserver, {native_parser:true, safe:true}) @@ -23,8 +19,8 @@ db.open( (err, db) -> exports.devicesCollection = collection ) - db.collection('profiles', (err, collection) -> - exports.profilesCollection = collection + db.collection('presets', (err, collection) -> + exports.presetsCollection = collection ) ) diff --git a/mongo-query.coffee b/mongo-query.coffee new file mode 100644 index 0000000..40a3b2a --- /dev/null +++ b/mongo-query.coffee @@ -0,0 +1,80 @@ +common = require './common' + + +test = (obj, query) -> + for k,v of query + if k.charAt(0) == '$' # this is a logical operator + res = testExpressions(obj, v, k) + else + value = common.getParamValueFromPath(obj, k) + + if common.typeOf(v) isnt common.OBJECT_TYPE + if common.typeOf(value) is common.ARRAY_TYPE + res = value.indexOf(v) != -1 + else + res = v == value + else + for k2,v2 of v + switch k2 + when '$ne' + if common.typeOf(value) is common.ARRAY_TYPE + res = value.indexOf(v2) == -1 + else + res = value != v2 + when '$lt' + res = value < v2 + when '$lte' + res = value <= v2 + when '$gt' + res = value > v2 + when '$gte' + res = value >= v2 + when '$regex' + res = v2.test(value) + when '$in' + throw new Error('Operator not supported') + when '$nin' + throw new Error('Operator not supported') + when '$all' + throw new Error('Operator not supported') + when '$exists' + throw new Error('Operator not supported') + else + throw new Error('Operator not supported') + + if not res + return false + + return true + + +testExpressions = (obj, expressions, lop) -> + for f in expressions + res = test(obj, f) + + switch lop + when '$and' + return false if not res + when '$or' + return true if res + when '$nor' + return false if res + + return switch lop + when '$and' then true + when '$or' then false + when '$nor' then true + else throw new Error('Unknown logical operator') + + +projection = (query, proj) -> + for k,v of query + if k.charAt(0) == '$' # this is a logical operator + for q in v + projection(q, proj) + else + proj[k] = 1 + + +exports.test = test +exports.projection = projection diff --git a/presets.coffee b/presets.coffee new file mode 100644 index 0000000..407e66f --- /dev/null +++ b/presets.coffee @@ -0,0 +1,115 @@ +config = require './config' +common = require './common' +db = require './db' +mongoQuery = require './mongo-query' + + +getPresets = (callback) -> + db.memcached.get('presets', (err, res) -> + throw new Error(err) if err + + if res + callback(res) + return + + db.presetsCollection.find().toArray((err, presets) -> + throw new Error(err) if err + + db.memcached.set('presets', presets, config.PRESETS_CACHE_DURATION) + callback(presets) + ) + ) + + +getPresetsHash = (callback) -> + db.memcached('presets_hash', (err, res) -> + if res + callback(res) + else + getPresets((presets) -> + hash = callback(calculatePresetsHash(presets)) + db.memcached.set('presets_hash', hash, config.PRESETS_CACHE_DURATION, (err, res) -> + callback(hash) + ) + ) + ) + + +calculatePresetsHash = (presets) -> + crypto = require('crypto') + hash = crypto.createHash('md5').update(JSON.stringify(presets)).digest('hex') + return hash + + +exports.assertPresets = (deviceId, presetsHash, callback) -> + getPresets((presets) -> + # only fetch relevant params + projection = {} + for p in presets + mongoQuery.projection(p.precondition, projection) + + for c in p.configurations + switch c.type + when 'value', 'age' + projection[c.name] = 1 + when 'firmware' + projection['InternetGatewayDevice.DeviceInfo.SoftwareVersion'] = 1 + else + throw new Error('Unknown configuration type') + + db.devicesCollection.findOne({'_id' : deviceId}, projection, (err, device) -> + devicePresets = [] + for p in presets + if mongoQuery.test(device, p.precondition) + devicePresets.push(p) + + configurations = accumulateConfigurations(devicePresets) + now = Date.now() + taskList = [] + expiry = config.PRESETS_CACHE_DURATION + getParameterValues = [] + setParameterValues = [] + for c in configurations + switch c.type + when 'value' + if c.value != common.getParamValueFromPath(device, "#{c.name}._value") + setParameterValues.push([c.name, c.value]) + when 'age' + timeDiff = (now - common.getParamValueFromPath(device, "#{c.name}._timestamp")) / 1000 + if (c.age - timeDiff < config.PRESETS_TIME_PADDING) + expiry = Math.min(expiry, c.age) + getParameterValues.push(c.name) + else + expiry = Math.min(expiry, c.age - timeDiff) + else + throw new Error('Unknown configuration type') + if getParameterValues.length + taskList.push {device : deviceId, name : 'getParameterValues', parameterNames: getParameterValues, timestamp : db.mongo.Timestamp()} + if setParameterValues.length + taskList.push {device : deviceId, name : 'setParameterValues', parameterValues: getParameterValues, timestamp : db.mongo.Timestamp()} + + if not presetsHash + presetsHash = calculatePresetsHash(presets) + db.memcached.set('presets_hash', presetsHash, expiry - config.PRESETS_TIME_PADDING, (err, res) -> + ) + + db.memcached.set("#{deviceId}_presets_hash", presetsHash, expiry - config.PRESETS_TIME_PADDING, (err, res) -> + callback(taskList) + ) + ) + ) + + +accumulateConfigurations = (presets) -> + maxWeights = {} + configurations = {} + for p in presets + for c in p.configurations + configurationHash = if c.name? then "#{c.type}_#{c.name}" else c.type + + if not maxWeights[configurationHash]? or p.weight > maxWeights[configurationHash] + configurations[configurationHash] = c + maxWeights[configurationHash] = p.weight + + configurationsList = (configurations[c] for c of configurations) + return configurationsList diff --git a/profiles.coffee b/profiles.coffee deleted file mode 100644 index f414987..0000000 --- a/profiles.coffee +++ /dev/null @@ -1,123 +0,0 @@ -config = require './config' -db = require './db' - - -getProfiles = (callback) -> - db.memcached.get('profiles', (err, res) -> - throw err if err - - if res - callback(res) - return - - db.profilesCollection.find().toArray((err, res) -> - if err - callback(err, res) - return - - profiles = {} - for p in res - profiles[p['_id']] = p - - db.memcached.set('profiles', profiles, config.PROFILES_CACHE_DURATION) - callback(profiles) - ) - ) - - -getProfilesHash = (callback) -> - db.memcached('profiles_hash', (err, res) -> - if res - callback(res) - else - getProfiles((profiles) -> - hash = callback(calculateProfilesHash(profiles)) - db.memcached.set('profiles_hash', hash, config.PROFILES_CACHE_DURATION, (err, res) -> - callback(hash) - ) - ) - ) - - -calculateProfilesHash = (profiles) -> - crypto = require('crypto') - hash = crypto.createHash('md5').update(JSON.stringify(profiles)).digest('hex') - return hash - - -exports.findDeviceConfigurationDiscrepancy = (deviceId, profilesHash, callback) -> - getProfiles((profiles) -> - params = {'_tags' : 1} - for p of profiles - profile = profiles[p] - for c in profile.configurations - if c.type == 'get' or c.type == 'set' - params[c.name] = 1 - - db.devicesCollection.findOne({'_id' : deviceId}, params, (err, device) -> - tags = device['_tags'] or [] - tags.push('default') - - deviceProfiles = [] - for t in tags - if profiles[t]? - deviceProfiles.push(profiles[t]) - - combinedConfigurations = combineProfileConfigurations(deviceProfiles) - now = Date.now() - taskList = [] - expiry = config.PROFILES_CACHE_DURATION - getParameterValues = [] - setParameterValues = [] - for c in combinedConfigurations - switch c.type - when 'set' - if c.value != getDeviceParamFromPath(device, "#{c.name}._value") - setParameterValues.push([c.name, c.value]) - when 'get' - timeDiff = (now - getDeviceParamFromPath(device, "#{c.name}._timestamp")) / 1000 - if (c.age - timeDiff < config.PROFILES_TIME_PADDING) - expiry = Math.min(expiry, c.age) - getParameterValues.push(c.name) - else - expiry = Math.min(expiry, c.age - timeDiff) - else - throw new Error('Unknown configuration type') - if getParameterValues.length - taskList.push {device : deviceId, name : 'getParameterValues', parameterNames: getParameterValues, timestamp : db.mongo.Timestamp()} - if setParameterValues.length - taskList.push {device : deviceId, name : 'setParameterValues', parameterValues: getParameterValues, timestamp : db.mongo.Timestamp()} - - if not profilesHash - profilesHash = calculateProfilesHash(profiles) - db.memcached.set('profiles_hash', profilesHash, expiry - config.PROFILES_TIME_PADDING, (err, res) -> - ) - - db.memcached.set("#{deviceId}_profiles_hash", profilesHash, expiry - config.PROFILES_TIME_PADDING, (err, res) -> - callback(taskList) - ) - ) - ) - - -combineProfileConfigurations = (profiles) -> - maxWeights = {} - configurations = {} - for p in profiles - for c in p.configurations - configurationHash = if c.name? then "#{c.type}_#{c.name}" else c.type - - if not maxWeights[configurationHash]? or p.weight > maxWeights[configurationHash] - configurations[configurationHash] = c - maxWeights[configurationHash] = p.weight - - configurationsList = (configurations[c] for c of configurations) - return configurationsList - - -getDeviceParamFromPath = (device, paramPath) -> - path = paramPath.split('.') - ref = device - for p in path - ref = ref[p] - return ref \ No newline at end of file