source: sage/dsage/database/monitordb.py @ 7660:62b7be191970

Revision 7660:62b7be191970, 10.0 KB checked in by Yi Qiang <yqiang@…>, 6 years ago (diff)

Refactored stats code.
Line 
1##############################################################################
2#                                                                     
3#  DSAGE: Distributed SAGE                     
4#                                                                             
5#       Copyright (C) 2006, 2007 Yi Qiang <yqiang@gmail.com>               
6#                                                                           
7#  Distributed under the terms of the GNU General Public License (GPL)       
8#
9#    This code is distributed in the hope that it will be useful,
10#    but WITHOUT ANY WARRANTY; without even the implied warranty of
11#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
12#    General Public License for more details.
13#
14#  The full text of the GPL is available at:
15#
16#                  http://www.gnu.org/licenses/
17#
18##############################################################################
19
20import datetime
21import os
22import sqlite3
23
24from twisted.python import log
25
26import sage.dsage.database.sql_functions as sql_functions
27from sage.dsage.misc.constants import DSAGE_DIR
28
29class MonitorDatabase(object):
30    """
31    This table keeps track of workers.
32   
33    """
34       
35    CREATE_MONITOR_TABLE = """CREATE TABLE monitors
36    (
37     uuid text NOT NULL UNIQUE,
38     hostname TEXT,
39     ip TEXT,
40     workers INTEGER,
41     sage_version text,
42     os text,
43     kernel_version TEXT,
44     cpus INTEGER,
45     cpu_speed INTEGER,
46     cpu_model TEXT,
47     mem_total INTEGER,
48     mem_free INTEGER,
49     connected BOOL,
50     busy BOOL,
51     anonymous BOOL DEFAULT 0,
52     last_connection timestamp
53    )
54    """
55   
56    def __init__(self, db_file=os.path.join(DSAGE_DIR, 'db', 'dsage.db'),
57                 log_file=os.path.join(DSAGE_DIR, 'server.log'), log_level=0,
58                 test=False):
59        self.tablename = 'monitors'
60        if test:
61            self.db_file = 'monitordb_test.db'
62            self.log_level = 5
63            self.log_file = 'monitordb_test.log'
64        else:
65            self.db_file = db_file
66            self.log_file = log_file
67            self.log_level = log_level
68            if not os.path.exists(self.db_file):
69                dir, file = os.path.split(self.db_file)
70                if not os.path.isdir(dir):
71                    os.mkdir(dir)
72        self.con = sqlite3.connect(self.db_file,
73                isolation_level=None,
74                detect_types=sqlite3.PARSE_DECLTYPES|sqlite3.PARSE_COLNAMES)
75        sql_functions.optimize_sqlite(self.con)
76        self.con.text_factory = str
77       
78        if sql_functions.table_exists(self.con, self.tablename) is None:
79            sql_functions.create_table(self.con, 
80                                       self.tablename,
81                                       self.CREATE_MONITOR_TABLE)
82            self.con.commit()
83       
84    def _set_parameter(self, uuid, key, value):
85        query = """UPDATE monitors
86        SET %s=?
87        WHERE uuid=?""" % (key)
88        cur = self.con.cursor()
89        cur.execute(query, (value, uuid))
90        self.con.commit()
91   
92    def set_anonymous(self, uuid, anonymous=True):
93        return self._set_parameter(uuid, 'anonymous', anonymous)
94       
95    def add_monitor(self, host_info):
96        query = """INSERT INTO monitors
97        (uuid,
98         hostname,
99         ip,
100         workers,
101         sage_version,
102         os,
103         kernel_version,
104         cpus,
105         cpu_speed,
106         cpu_model,
107         mem_total,
108         mem_free)
109        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
110        """
111       
112        uuid = host_info['uuid']
113        hostname = host_info['hostname']
114        ip = host_info['ip']
115        workers = host_info['workers']
116        sage_version = host_info['sage_version']
117        os = host_info['os']
118        kernel_version = host_info['kernel_version']
119        cpus = host_info['cpus']
120        cpu_speed = host_info['cpu_speed']
121        cpu_model = host_info['cpu_model']
122        mem_total = host_info['mem_total']
123        mem_free = host_info['mem_free']
124       
125        cur = self.con.cursor()
126        cur.execute(query, (uuid, hostname, ip, workers, sage_version, os, 
127                            kernel_version, cpus, cpu_speed, cpu_model,
128                            mem_total, mem_free))
129        self.con.commit()
130       
131    def update_monitor(self, host_info):
132        query = """UPDATE monitors
133        SET hostname = ?, ip = ?, workers = ?, sage_version = ?, os = ?,
134        kernel_version = ?, cpus = ?, cpu_speed = ?, cpu_model = ?, mem_total
135        = ?, mem_free = ?
136        WHERE uuid = ?
137        """
138       
139        uuid = host_info['uuid']
140        hostname = host_info['hostname']
141        ip = host_info['ip']
142        workers = host_info['workers']
143        sage_version = host_info['sage_version']
144        os = host_info['os']
145        kernel_version = host_info['kernel_version']
146        cpus = host_info['cpus']
147        cpu_speed = host_info['cpu_speed']
148        cpu_model = host_info['cpu_model']
149        mem_total = host_info['mem_total']
150        mem_free = host_info['mem_free']
151
152        cur = self.con.cursor()
153        cur.execute(query, (hostname, ip, workers, sage_version, os, 
154                            kernel_version, cpus, cpu_speed, cpu_model,
155                            mem_total, mem_free, uuid))
156       
157    def get_monitor(self, uuid):
158        query = """SELECT
159        uuid,
160        workers,
161        hostname,
162        ip,
163        anonymous,
164        sage_version,
165        os
166        FROM monitors
167        WHERE uuid = ?"""
168       
169        cur = self.con.cursor()
170        cur.execute(query, (uuid,))
171        result = cur.fetchone()
172        if result is None:
173            return result
174        columns = [desc[0] for desc in cur.description]
175        monitor = dict(zip(columns, result))
176        for k, v in monitor.iteritems():
177            if k == 'anonymous':
178                monitor[k] = bool(v)
179               
180        return monitor
181
182    def get_monitor_list(self):
183        """
184        Returns a list of connected monitors.
185       
186        """
187       
188        query = """SELECT uuid, hostname, ip, anonymous, sage_version, os
189                   FROM monitors
190                   WHERE connected"""
191        cur = self.con.cursor()
192        cur.execute(query)
193        result = cur.fetchall()
194        columns = [desc[0] for desc in cur.description]
195        monitors = [dict(zip(columns, monitor)) for monitor in result]
196        for monitor in monitors:
197            for k, v in monitor.iteritems():
198                if k == 'anonymous':
199                    monitor[k] = bool(v)
200                   
201        return monitors
202       
203    def set_connected(self, uuid, connected=True):
204        """
205        Sets the connected status of a monitor.
206       
207        Parameters:
208        uuid -- string
209        connected -- bool
210       
211        """
212       
213        cur = self.con.cursor()
214        if connected:
215            query = """UPDATE monitors SET connected=1, last_connection=?
216            WHERE uuid=?"""
217            cur.execute(query, (datetime.datetime.now(), uuid))
218        else:
219            query = """UPDATE monitors SET connected=0 WHERE uuid=?"""
220            cur.execute(query, (uuid,))
221           
222        self.con.commit()
223   
224    def is_connected(self, uuid):
225        """
226        Returns whether the monitor is connected.
227       
228        """
229       
230        query = """SELECT connected FROM monitors WHERE uuid = ?"""
231        cur = self.con.cursor()
232        cur.execute(query, (uuid,))
233        result = cur.fetchone()[0]
234       
235        return result
236       
237    def set_busy(self, uuid, busy):
238        """
239        Sets whether or not a worker is doing a job.
240       
241        """
242       
243        if busy:
244            query = """UPDATE monitors SET busy=1 WHERE uuid=?"""
245        else:
246            query = """UPDATE monitors SET busy=0 WHERE uuid=?"""
247       
248        cur = self.con.cursor()
249        cur.execute(query, (uuid,))
250        self.con.commit()
251       
252    def get_worker_count(self, connected, busy=False):
253        """
254        Returns the number of workers.
255       
256        Parameters:
257        connected -- bool
258        busy -- bool
259       
260        """
261                   
262        if connected and not busy:
263            query = """SELECT workers FROM monitors WHERE connected AND NOT busy"""
264        elif connected and busy:
265            query = """SELECT workers FROM monitors WHERE connected AND busy"""
266        elif not connected and not busy:
267            query = """SELECT workers FROM monitors WHERE NOT connected AND NOT busy"""
268        elif not connected and busy:
269            query = """SELECT workers FROM monitors WHERE NOT connected AND busy"""
270        cur = self.con.cursor()
271        cur.execute(query)
272           
273        result = cur.fetchall()
274       
275        return sum(w[0] for w in result)
276
277    def get_cpu_speed(self, connected=True, busy=False):
278        """
279        Returns the aggregate cpu speed in Mhz.
280       
281        Parameters:
282        connected -- bool
283       
284        """
285       
286        if connected and busy:
287            query = """SELECT cpu_speed, workers FROM monitors
288            WHERE connected AND busy"""
289        elif connected:
290            query = """SELECT cpu_speed, workers FROM monitors
291            WHERE connected"""
292        else:
293            query = """SELECT cpu_speed, workers FROM monitors"""
294           
295        cur = self.con.cursor()
296        cur.execute(query)
297       
298        result = cur.fetchall()
299       
300        cpu_speed = sum([s[0]*s[1] for s in result])
301       
302        return cpu_speed
303   
304    def get_cpu_count(self, connected=True):
305        """
306        Returns the number of cpus that are available.
307       
308        Parameters:
309        connected -- bool
310       
311        """
312       
313        if connected:
314            query = """SELECT workers, cpus FROM monitors WHERE connected"""
315        else:
316            query = """SELECT workers, cpus FROM monitors"""
317       
318        cur = self.con.cursor()
319        cur.execute(query)
320       
321        result = cur.fetchall()
322       
323        cpu_count = sum(min(s[0:2]) for s in result)
324       
325        return cpu_count
Note: See TracBrowser for help on using the repository browser.