common, live, webui: Add very basic data view page for live databases

Author: justinclift

common/live.go

@@ -64,11 +64,11 @@ type LiveDBQueryResponse struct {
 
 // LiveDBRequest holds the fields used for sending requests to our AMQP backend
 type LiveDBRequest struct {
-	Operation      string `json:"operation"`
-	DBOwner        string `json:"dbowner"`
-	DBName         string `json:"dbname"`
-	Query          string `json:"query"`
-	RequestingUser string `json:"requesting_user"`
+	Operation      string      `json:"operation"`
+	DBOwner        string      `json:"dbowner"`
+	DBName         string      `json:"dbname"`
+	Data           interface{} `json:"data,omitempty"`
+	RequestingUser string      `json:"requesting_user"`
 }
 
 // LiveDBResponse holds the fields used for receiving (non-query) responses from our AMQP backend
@@ -78,6 +78,26 @@ type LiveDBResponse struct {
 	Error  string `json:"error"`
 }
 
+// LiveDBRowsRequest holds the data used when making an AMQP rows request
+type LiveDBRowsRequest struct {
+	DbTable   string `json:"db_table"`
+	SortCol   string `json:"sort_col"`
+	SortDir   string `json:"sort_dir"`
+	CommitID  string `json:"commit_id"`
+	RowOffset int    `json:"row_offset"`
+	MaxRows   int    `json:"max_rows"`
+}
+
+// LiveDBRowsResponse holds the fields used for receiving database page row responses from our AMQP backend
+type LiveDBRowsResponse struct {
+	Node         string          `json:"node"`
+	DatabaseSize int64           `json:"database_size"`
+	DefaultTable string          `json:"default_table"`
+	RowData      SQLiteRecordSet `json:"row_data"`
+	Tables       []string        `json:"tables"`
+	Error        string          `json:"error"`
+}
+
 // LiveDBs is used for general purpose holding of details about live databases
 type LiveDBs struct {
 	DBOwner     string    `json:"owner_name"`
@@ -292,7 +312,7 @@ func MQCreateResponse(msg amqp.Delivery, channel *amqp.Channel, nodeName, result
 }
 
 // MQRequest is the main function used for sending requests to our AMQP backend
-func MQRequest(channel *amqp.Channel, queue, operation, requestingUser, dbOwner, dbName, query string) (result []byte, err error) {
+func MQRequest(channel *amqp.Channel, queue, operation, requestingUser, dbOwner, dbName string, data interface{}) (result []byte, err error) {
 	// Create a temporary AMQP queue for receiving the response
 	var q amqp.Queue
 	q, err = channel.QueueDeclare("", false, false, true, false, nil)
@@ -305,7 +325,7 @@ func MQRequest(channel *amqp.Channel, queue, operation, requestingUser, dbOwner,
 		Operation:      operation,
 		DBOwner:        dbOwner,
 		DBName:         dbName,
-		Query:          query,
+		Data:           data,
 		RequestingUser: requestingUser,
 	}
 	var z []byte

common/memcache.go

@@ -229,7 +229,7 @@ func SetUserStatusUpdates(userName string, numUpdates int) error {
 	return nil
 }
 
-// TableRowsCacheKey generates a predictable cache key for SQLite row data
+// TableRowsCacheKey generates a predictable cache key for SQLite row data.  ONLY for standard databases
 func TableRowsCacheKey(prefix string, loggedInUser string, dbOwner string, dbFolder string, dbName string, commitID string, dbTable string, rows int) string {
 	var cacheString string
 	if strings.ToLower(loggedInUser) == strings.ToLower(dbOwner) {

common/sqlite.go

@@ -1142,6 +1142,135 @@ func SQLiteSanityCheck(fileName string) (tables []string, err error) {
 	return
 }
 
+// SQLiteReadDatabasePage opens a SQLite database (locally) and returns a "page" of rows from it, for display in the
+// database view page.  Note that the dbSize return value is only set for live databases.
+func SQLiteReadDatabasePage(bucket, id, loggedInUser, dbOwner, dbName, dbTable, sortCol, sortDir, commitID string, rowOffset, maxRows int, isLive bool) (tables []string, defaultTable string, rowData SQLiteRecordSet, dbSize int64, err error) {
+	dbFolder := "/"
+
+	// Get a handle from Minio for the database object
+	var sdb *sqlite.Conn
+	if isLive {
+		// Open live database file
+		sdb, err = OpenSQLiteDatabaseLive(Conf.Live.StorageDir, dbOwner, dbName)
+		if err != nil {
+			return
+		}
+
+		// We also return the file size for live database files
+		var z os.FileInfo
+		z, err = os.Stat(filepath.Join(Conf.Live.StorageDir, dbOwner, dbName, "live.sqlite"))
+		if err != nil {
+			return
+		}
+		dbSize = z.Size()
+	} else {
+		// Open standard database
+		sdb, err = OpenSQLiteDatabase(bucket, id)
+		if err != nil {
+			return
+		}
+	}
+	defer sdb.Close()
+
+	// Retrieve the list of tables and views in the database
+	tables, err = TablesAndViews(sdb, dbName)
+	if err != nil {
+		return
+	}
+
+	// If a specific table or view was requested, check that it's present
+	if dbTable != "" {
+		tablePresent := false
+		for _, tbl := range tables {
+			if tbl == dbTable {
+				tablePresent = true
+			}
+		}
+		if tablePresent == false {
+			// The requested table or view doesn't exist in the database, so pick one of the tables that does
+			for _, t := range tables {
+				err = ValidatePGTable(t)
+				if err == nil {
+					// Validation passed, so use this table or view
+					dbTable = t
+					defaultTable = t
+					break
+				}
+			}
+		}
+	}
+
+	// If a specific table wasn't requested, use the first table in the database that passes validation
+	if dbTable == "" {
+		for _, i := range tables {
+			if i != "" {
+				err = ValidatePGTable(i)
+				if err == nil {
+					// The database table name is acceptable, so use it
+					dbTable = i
+					break
+				}
+			}
+		}
+	}
+
+	// If a sort column was requested, verify it exists
+	if sortCol != "" {
+		var colList []sqlite.Column
+		colList, err = sdb.Columns("", dbTable)
+		if err != nil {
+			log.Printf("Error when reading column names for table '%s': %v\n", SanitiseLogString(dbTable), err.Error())
+			return
+		}
+		colExists := false
+		for _, j := range colList {
+			if j.Name == sortCol {
+				colExists = true
+			}
+		}
+		if colExists == false {
+			// The requested sort column doesn't exist, so we fall back to no sorting
+			sortCol = ""
+		}
+	}
+
+	// Validate the table name, just to be careful
+	if dbTable != "" {
+		err = ValidatePGTable(dbTable)
+		if err != nil {
+			// Validation failed, so don't pass on the table name
+
+			// If the failed table name is "{{ db.Tablename }}", don't bother logging it.  It's just a search
+			// bot picking up AngularJS in a string and doing a request with it
+			if dbTable != "{{ db.Tablename }}" {
+				log.Printf("Validation failed for table name: '%s': %s", SanitiseLogString(dbTable), err)
+			}
+			return
+		}
+	}
+
+	var okCache bool
+	if !isLive {
+		rowCacheKey := TableRowsCacheKey(fmt.Sprintf("tablejson/%s/%s/%d", sortCol, sortDir, rowOffset),
+			loggedInUser, dbOwner, dbFolder, dbName, commitID, dbTable, maxRows)
+		okCache, err = GetCachedData(rowCacheKey, &rowData)
+		if err != nil {
+			log.Printf("Error retrieving page data from cache: %v", err)
+		}
+	}
+
+	// If the row data wasn't in cache, read it from the database
+	if !okCache {
+		rowData, err = ReadSQLiteDB(sdb, dbTable, sortCol, sortDir, maxRows, rowOffset)
+		if err != nil {
+			// Some kind of error when reading the database data
+			return
+		}
+		rowData.Tablename = dbTable
+	}
+	return
+}
+
 // SQLiteRunQuery runs a SQLite query.  DO NOT use this for user provided SQL queries.  For those,
 // use SQLiteRunQueryDefensive().
 func SQLiteRunQuery(sdb *sqlite.Conn, querySource QuerySource, dbQuery string, ignoreBinary, ignoreNull bool) (memUsed, memHighWater int64, dataRows SQLiteRecordSet, err error) {

live/main.go

@@ -10,6 +10,7 @@ package main
 import (
 	"encoding/json"
 	"errors"
+	"fmt"
 	"log"
 	"os"
 	"path/filepath"
@@ -142,7 +143,7 @@ func main() {
 			}
 
 			if com.AmqpDebug {
-				log.Printf("Decoded request on '%s'.  Correlation ID: '%s', request operation: '%s', request query: '%s'", com.Conf.Live.Nodename, msg.CorrelationId, req.Operation, req.Query)
+				log.Printf("Decoded request on '%s'.  Correlation ID: '%s', request operation: '%s', request query: '%v'", com.Conf.Live.Nodename, msg.CorrelationId, req.Operation, req.Data)
 			}
 
 			// Handle each operation
@@ -169,7 +170,7 @@ func main() {
 			case "columns":
 				var columns []sqlite.Column
 				var errCode com.AMQPErrorCode
-				columns, err, errCode = com.SQLiteGetColumnsLive(com.Conf.Live.StorageDir, req.DBOwner, req.DBName, req.Query) // We use the req.Query field to pass the table name
+				columns, err, errCode = com.SQLiteGetColumnsLive(com.Conf.Live.StorageDir, req.DBOwner, req.DBName, fmt.Sprintf("%s", req.Data))
 				if err != nil {
 					resp := com.LiveDBColumnsResponse{Node: com.Conf.Live.Nodename, Columns: []sqlite.Column{}, Error: err.Error(), ErrCode: errCode}
 					err = com.MQResponse("COLUMNS", msg, ch, com.Conf.Live.Nodename, resp)
@@ -210,7 +211,7 @@ func main() {
 			case "execute":
 				// Execute a SQL statement on the database file
 				var rowsChanged int
-				rowsChanged, err = com.SQLiteExecuteQueryLive(com.Conf.Live.StorageDir, req.DBOwner, req.DBName, req.RequestingUser, req.Query)
+				rowsChanged, err = com.SQLiteExecuteQueryLive(com.Conf.Live.StorageDir, req.DBOwner, req.DBName, req.RequestingUser, fmt.Sprintf("%s", req.Data))
 				if err != nil {
 					resp := com.LiveDBExecuteResponse{Node: com.Conf.Live.Nodename, RowsChanged: 0, Error: err.Error()}
 					err = com.MQResponse("EXECUTE", msg, ch, com.Conf.Live.Nodename, resp)
@@ -250,7 +251,7 @@ func main() {
 
 			case "query":
 				var rows com.SQLiteRecordSet
-				rows, err = com.SQLiteRunQueryLive(com.Conf.Live.StorageDir, req.DBOwner, req.DBName, req.RequestingUser, req.Query)
+				rows, err = com.SQLiteRunQueryLive(com.Conf.Live.StorageDir, req.DBOwner, req.DBName, req.RequestingUser, fmt.Sprintf("%s", req.Data))
 				if err != nil {
 					resp := com.LiveDBQueryResponse{Node: com.Conf.Live.Nodename, Results: com.SQLiteRecordSet{}, Error: err.Error()}
 					err = com.MQResponse("QUERY", msg, ch, com.Conf.Live.Nodename, resp)
@@ -268,6 +269,38 @@ func main() {
 				}
 				continue
 
+			case "rowdata":
+				// Extract the request information
+				// FIXME: Add type checks for safety instead of blind coercing
+				var reqData = make(map[string]interface{})
+				reqData = req.Data.(map[string]interface{})
+				dbTable := reqData["db_table"].(string)
+				sortCol := reqData["sort_col"].(string)
+				sortDir := reqData["sort_dir"].(string)
+				commitID := reqData["commit_id"].(string)
+				maxRows := int(reqData["max_rows"].(float64))
+				rowOffset := int(reqData["row_offset"].(float64))
+
+				// Open the SQLite database and read the row data
+				resp := com.LiveDBRowsResponse{Node: com.Conf.Live.Nodename, RowData: com.SQLiteRecordSet{}}
+				resp.Tables, resp.DefaultTable, resp.RowData, resp.DatabaseSize, err =
+					com.SQLiteReadDatabasePage(fmt.Sprintf("live-%s", req.DBOwner), req.DBName, req.RequestingUser, req.DBOwner, req.DBName, dbTable, sortCol, sortDir, commitID, rowOffset, maxRows, true)
+				if err != nil {
+					resp := com.LiveDBErrorResponse{Node: com.Conf.Live.Nodename, Error: err.Error()}
+					err = com.MQResponse("ROWDATA", msg, ch, com.Conf.Live.Nodename, resp)
+					if err != nil {
+						log.Printf("Error: occurred on '%s' in MQResponse() while constructing an AMQP error message response: '%s'", com.Conf.Live.Nodename, err)
+					}
+					continue
+				}
+
+				// Return the row data to the caller
+				err = com.MQResponse("ROWDATA", msg, ch, com.Conf.Live.Nodename, resp)
+				if err != nil {
+					log.Printf("Error: occurred on '%s' in MQResponse() while constructing the AMQP query response: '%s'", com.Conf.Live.Nodename, err)
+				}
+				continue
+
 			case "tables":
 				var tables []string
 				tables, err = com.SQLiteGetTablesLive(com.Conf.Live.StorageDir, req.DBOwner, req.DBName)