External File manager prototype

[pavlis]

Rather than immediately push this as a new branch for inclusion the master branch I think we might want to discuss if what I've got below is a good idea or something that will just cause us headaches down the road.

This came up in writing the CRUD section when I was thinking about the D operation. Deletes are problematic when the waveform data are stored in files for a whole list of reasons. A much longer discussion of the issues can be found in our google doc on the database api design.

I developed the following prototype for something I called an ExternalFileManager class. Read the docstring comments to hopefully understand the concept and how it works. The first code block contains the class definition for this thing. The second is a very crude test script discussed at the bottom of this discussion page:

import os
from bson.objectid import ObjectId
from mspasspy.ccore.utility import MsPASSError
class ExternalFileManager:
    """
    When storing waveform data as external files in any format it can 
    be problematic to know what files need to be deleted.  Becomes 
    especially true as the size of a data set increases an data are 
    mixed in multiple files assembled in some directory structure that 
    cannot be handled with traditional directory based file management. 
    For example, if one was doing station based processing but the data 
    were stored in common source gathers it would be hard to know which 
    files could be safely deleted. 
    
    This class is designed for use on MongoDB waveform collections 
    for which the data are stored in files instead of gridfs.  It is 
    intended for use by specialized cleanup scripts for clearing files 
    that are not being used.  This is most common during preprocessing 
    to assemble a working data set from an import format like miniseed. 
    Use of this class is best limited to a four step procedure:
        1.  Create an instance of the class pointing at the waveform 
            collection to be managed.   
        2.  Run the workflows the data stored in a set of files
            defined in the waveform collection used to create this class.
        3.  Either within the workflow or in separate cleanup script 
            call the delete method for each waveform that is will no 
            longer be needed.
        4.  Call the flush method when finished (Note the destructor 
            for the class will do this automatically if you fail to do 
            so.  Beware as that step can cause files to mysteriously 
            disappear.)
        
    We note that calling the flush algorithm automatically 
    in the destructor is somewhat intrisically dangerous but essential 
    to avoid stranding unreferenced files.  Calls to delete remove the 
    documents that reference each independent data object so flush 
    will be essential to remove the sample data as well as the metadata. 
    
    The algorithm used used two keys to define file names:  one to hold a 
    directory path string and a second to hold the file name at the leaf 
    node of a directory tree.  The default keys for these two concepts
    are "dir"  and "dfile" respectively.   The keys can be changed 
    if necessary with options in the constructor.  
    """
    def __init__(self, db=None,collection="import_miniseed_ensemble",
                 dirkey='dir',dfilekey='dfile'):
        """
        The constructor requires a mspasspy Database handle.  It initializes
        an internal set container of 2 element tuples that the delete 
        method hits for each transaction.   The set container is initialized
        here to an empty set. The tuple contents are filled in delete with 
        attributes for the directory and file name
        (default keys "dir" and "dfile" respectively)
        
        :param db:  mspasspy Database handle to be managed.
        :param collection:  string defining the collection name referencing 
           external files to be managed. (default 'import_miniseed_ensemble')
        :param dirkey: key used for directory path string of files
           defined in the collection being managed
        :param dfilekey:  key used for the file (leaf node) of field 
           defining a file to be managed. 
        """
        if db==None:
            raise MsPASSError('ExternalFileManger constructor: '
                        + 'Missing required arg 0=Database handle','Fatal')
        self.db=db
        self.dbwf=db[collection]
        self.collection=collection
        self.stage_for_delete=set()
        self.dfile_key=dfilekey
        self.dir_key=dirkey
        
    def delete(self,id=None,cursor=None,verbose=False):
        """
        This method does three things.  First, for each MongoDB document 
        it is asked to process it fetches the dir and dfile fields 
        (note the constructor can change the default keys).   ISecond, 
        it creates a 2 element tuple of dir,dfile and posts it to the 
        internal set container.  A set container has the property that 
        duplicate tuples will simply replace the previous content with 
        the same valuies.  Finally, the document referenced is deleted 
        from the collection being managed.  
        
        When the manager is deleted or the flush method is called the 
        unique tuples with no residual documents in the collection will
        cause files to be deleted.  (see flush method docstring)
        
        This method can support either a single objectid as input or 
        a cursor from a query.  The cursor version is more efficient 
        calling by id requires a special query for the document linked to 
        that id.  Note you must specify either the id or cursor parameter
        or the method will throw a MsPASSError exception.
        
        :param id:  If you need to delete one waveform document only use 
          this parameter to specify the id of the doc in the managed collection.
          Default is None.  
        :param cursor:  Use this parameter to pass a cursor to the managed 
          collection that defines the waveform documents to be managed.  
          Default is None.  
        :param verbose:  If true will write informational messages. When 
          false (the default) it works silentely. 

        """
        if id==None and cursor==None:
            raise MsPASSError("ExternalFileManager.delete method:  "
            + "usage - must specify a value for either id or cursor","Fatal")
        if id!=None:
            if not isinstance(id,ObjectId):
                raise MsPASSError("ExternalFileManager.delete method:  "
                        + 'id parameter received is not an ObjectId')
            query = {'_id' : id}
            delcursor=self.dbwf.find(query)
        else:
            delcursor=cursor
        
        if verbose:
            # Warning I think this is a potentially expensive operation if 
            # the cursor is large
            n_to_delete=len(list(delcursor))
            print('delete method is about to delete ',n_to_delete, 
                  ' documents in ',self.collection)
            delcursor.rewind()
        # Initially skip any document whe
        for doc in delcursor:
            if self.dfile_key in doc:
                if self.dir_key in doc:
                    dir=doc[self.dir_key]
                    dfile=doc[self.dfile_key]
                    tup=tuple([dir,dfile])
                    self.stage_for_delete.add(tup)
                    self.dbwf.delete_one({'_id' : doc['_id']})
                    
    def flush(self,verbose=False):
        """
        Use this method to clear (flush) any files unreferenced in the 
        database.  The method queries the database for any documents matching 
        all unique combinations of dir/dfile found in previous calls to 
        delete. When there is no match in the db it constructs a path using dir/dfile
        and attempts to delete the file.  
    
        :param verbose:  when true will print a message for each 
         file deleted.   By default (false) it does its work silently.

        :return:  list of file names that were not deleted because  
        the file constructed from dir/dfile did not exist.
        We use os.remove that we assume will throw an exception if 
        the file could not be removed at all for some reason.  
        This should perhaps be in handled with the appropriate
        exception handler.  Note an empty return means all file
    

        """
        failures=[]
        for pair in self.stage_for_delete:
           query={
               self.dir_key : pair[0],
               self.dfile_key : pair[1]
               }
           ndocs=self.dbwf.count_documents(query)
           if ndocs==0:
               fname=pair[0]+'/'+pair[1]
               if os.path.exists(fname):
                   os.remove(fname)
                   print('Deleted file=',fname)
               else:
                   failures.append(fname)
                   print('WARNING:  file=',fname,' does not appear to exist\n',
                    'Check for a relative path problem or a collision with another instance running in parallel')
        self.stage_for_delete.clear()
        return failures

    def __del__(self):
        """
        We require a destructor in this class because if the class goes out
        of scope it needs to call the method to delete any files that 
        no longer have any docs in the parent collection.  Without this 
        files could easily be stranded if someone accidentally deleted an
        instance of the class in a job.
        """
        print("Auto-deleting unreferenced files in ExternalFileManager destructor")
        failures=self.flush()
        if len(failures)>0:
            print('ExternalFileManager destructor auto-called flush method\n',
                 'The following files were not found that it tried to delete')
            for f in failures:
                print(f)

and here is the very crude test program to show it works:

import os
from ExternalFileManager import ExternalFileManager
from mspasspy.db.database import Database
from mspasspy.db.client import Client as DBClient
dbclient=DBClient()
db=Database(dbclient,'testfilemanager')
manager=ExternalFileManager(db,collection='wf_TimeSeries')
# Done for this test to assure we start with a blank slate
db.drop_collection('wf_TimeSeries')
# first create a set of empty files with dir and dfile entries 
# in a wf_TimeSeries collection
dir='testdir'
dfile='fileA'
path=dir+"/"+dfile
f=open(path,'w')
f.write('this is file A\n')
f.close()
doc=dict()
doc['dir']=dir
doc['dfile']=dfile
# Add file A three times
db.wf_TimeSeries.insert_one(doc)
# mongo adds the id with an insert so we have to clear it here
doc.pop('_id')
db.wf_TimeSeries.insert_one(doc)
doc.pop('_id')
db.wf_TimeSeries.insert_one(doc)
doc.pop('_id')
dfile='fileB'
path=dir+"/"+dfile
f=open(path,'w')
f.write('this is file B\n')
f.close()
doc['dfile']=dfile
# Add file B two times
db.wf_TimeSeries.insert_one(doc)
doc.pop('_id')
db.wf_TimeSeries.insert_one(doc)
print('done creating test data.')
n=db.wf_TimeSeries.count_documents({})
print('wf collection document count=',n)
# simple test clears all. 
curs=db.wf_TimeSeries.find({}).limit(5)
manager.delete(cursor=curs)
n=db.wf_TimeSeries.count_documents({})
print('wf collection document count after running manager=',n)
#manager.flush()
del manager

This thing is very crude. You have to create the directory "testdir" from the place you copy this code. It runs the code to manage two files "fileA" and "fileB" with 5 documents referencing them. If run in the form above all 5 documents will be deleted and the thing properly removes both fileA and fileB. If you change the llimit(5) to some smaller number like limit(4) only one of the files will be deleted. If you use limit(1) or limit(2) neither file will be deleted but the document count will be nonzero. That is what it was designed to do.

Anyway, I think this could be a really nice feature for managing some kinds of data. I made the default collection the one I used for the usarray data download where this will prove useful for storage management.