alfresco-community-repo/source/java/org/alfresco/repo/usage/UserUsageTrackingComponent.java

/*
 * Copyright (C) 2005-2009 Alfresco Software Limited.
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.

 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.

 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.

 * As a special exception to the terms and conditions of version 2.0 of 
 * the GPL, you may redistribute this Program in connection with Free/Libre 
 * and Open Source Software ("FLOSS") applications as described in Alfresco's 
 * FLOSS exception.  You should have recieved a copy of the text describing 
 * the FLOSS exception, and it is also available here: 
 * http://www.alfresco.com/legal/licensing"
 */
package org.alfresco.repo.usage;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.alfresco.model.ContentModel;
import org.alfresco.repo.node.db.NodeDaoService;
import org.alfresco.repo.node.db.NodeDaoService.ObjectArrayQueryCallback;
import org.alfresco.repo.security.authentication.AuthenticationUtil;
import org.alfresco.repo.security.authentication.AuthenticationUtil.RunAsWork;
import org.alfresco.repo.tenant.Tenant;
import org.alfresco.repo.tenant.TenantAdminService;
import org.alfresco.repo.tenant.TenantService;
import org.alfresco.repo.transaction.TransactionServiceImpl;
import org.alfresco.repo.transaction.RetryingTransactionHelper.RetryingTransactionCallback;
import org.alfresco.service.cmr.repository.ContentData;
import org.alfresco.service.cmr.repository.NodeRef;
import org.alfresco.service.cmr.repository.NodeService;
import org.alfresco.service.cmr.repository.StoreRef;
import org.alfresco.service.cmr.usage.UsageService;
import org.alfresco.service.namespace.QName;
import org.alfresco.util.Pair;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * User Usage Tracking Component - to allow user usages to be collapsed or re-calculated
 * 
 * - used by UserUsageCollapseJob to collapse usage deltas.
 * - used by UserUsageBootstrapJob to either clear all usages or (re-)calculate all missing usages.
 */
public class UserUsageTrackingComponent
{
    private static Log logger = LogFactory.getLog(UserUsageTrackingComponent.class);
    
    private TransactionServiceImpl transactionService;
    private ContentUsageImpl contentUsageImpl;
    
    private NodeService nodeService;
    private NodeDaoService nodeDaoService;
    private UsageService usageService;
    private TenantAdminService tenantAdminService;
    private TenantService tenantService;
    
    private StoreRef personStoreRef;
    
    private int clearBatchSize = 50;
    private int updateBatchSize = 50;
    
    private boolean enabled = true;
    private Lock writeLock = new ReentrantLock();
    
    public void setTransactionService(TransactionServiceImpl transactionService)
    {
        this.transactionService = transactionService;
    }
    
    public void setContentUsageImpl(ContentUsageImpl contentUsageImpl)
    {
        this.contentUsageImpl = contentUsageImpl;
    }
    
    public void setPersonStoreUrl(String storeUrl)
    {
        this.personStoreRef = new StoreRef(storeUrl);
    }
    
    public void setNodeService(NodeService nodeService)
    {
        this.nodeService = nodeService;
    }
    
    public void setNodeDaoService(NodeDaoService nodeDaoService)
    {
        this.nodeDaoService = nodeDaoService;
    }
    
    public void setUsageService(UsageService usageService)
    {
        this.usageService = usageService;
    }
    
    public void setTenantAdminService(TenantAdminService tenantAdminService)
    {
        this.tenantAdminService = tenantAdminService;
    }
    
    public void setTenantService(TenantService tenantService)
    {
        this.tenantService = tenantService;
    }
    
    public void setClearBatchSize(int clearBatchSize)
    {
        this.clearBatchSize = clearBatchSize;
    }
    
    public void setUpdateBatchSize(int updateBatchSize)
    {
        this.updateBatchSize = updateBatchSize;
    }
    
    public void setEnabled(boolean enabled)
    {
        this.enabled = enabled;
    }
    
    public void execute()
    {
        if (enabled == false || transactionService.isReadOnly())
        {
            return;
        }
        
        boolean locked = writeLock.tryLock();
        if (locked)
        {
            // collapse usages - note: for MT environment, will collapse for all tenants
            try
            {
                collapseUsages();
            }
            finally
            {
                writeLock.unlock();
            }
        }
    }
    
    // called once on startup
    public void bootstrap()
    {
        // default domain
        bootstrapInternal();
        
        if (tenantAdminService.isEnabled())
        {
            List<Tenant> tenants = tenantAdminService.getAllTenants();
            for (Tenant tenant : tenants)
            {
                AuthenticationUtil.runAs(new RunAsWork<Object>()
                {
                    public Object doWork() throws Exception
                    {
                        bootstrapInternal();
                        return null;
                    }
                }, tenantAdminService.getDomainUser(AuthenticationUtil.getSystemUserName(), tenant.getTenantDomain()));
            }
       }
    }
    
    public void bootstrapInternal()
    {
        if (transactionService.isReadOnly())
        {
            return;
        }
        
        boolean locked = writeLock.tryLock();
        if (locked)
        {
            try
            {
                if (enabled)
                {
                    // enabled - calculate missing usages
                    calculateMissingUsages();
                }
                else
                {
                    if (clearBatchSize != 0)
                    {
                        // disabled - remove all usages
                        clearAllUsages();
                    }
                }
            }
            finally
            {
                writeLock.unlock();
            }
        }
    }
    
    /**
     * Clear content usage for all users that have a usage.
     */
    private void clearAllUsages()
    {
        if (logger.isInfoEnabled()) 
        {
            logger.info("Disabled - clear non-missing user usages ...");
        }
        
        final Map<String, NodeRef> users = new HashMap<String, NodeRef>();
        
        RetryingTransactionCallback<Object> getUsersWithUsage = new RetryingTransactionCallback<Object>()
        {
            public Object execute() throws Throwable
            {
                // get people (users) with calculated usage
                ObjectArrayQueryCallback userHandler = new ObjectArrayQueryCallback()
                {
                    public boolean handle(Object[] arr)
                    {
                        String username = (String)arr[0];
                        String uuid = (String)arr[1];
                        
                        users.put(username, new NodeRef(personStoreRef, uuid));
                        
                        return true; // continue to next node (more required)
                    }
                };
                nodeDaoService.getUsersWithUsage(personStoreRef, userHandler);
                
                return null;
            }
        };
        
        // execute in READ-ONLY txn
        transactionService.getRetryingTransactionHelper().doInTransaction(getUsersWithUsage, true);
        
        if (logger.isInfoEnabled()) 
        {
            logger.info("Found " + users.size() + " users to clear");
        }
        
        int clearCount = 0;
        int batchCount = 0;
        int totalCount = 0;
        
        List<NodeRef> batchPersonRefs = new ArrayList<NodeRef>(clearBatchSize);
        for (NodeRef personNodeRef : users.values())
        {
            batchPersonRefs.add(personNodeRef);
            batchCount++;
            totalCount++;
            
            if ((batchCount == clearBatchSize) || (totalCount == users.size()))
            {
                int cleared = clearUsages(batchPersonRefs);
                clearCount = clearCount + cleared;
                
                batchPersonRefs.clear();
                batchCount = 0;
            }
        }
        
        if (logger.isInfoEnabled()) 
        {
            logger.info("... cleared non-missing usages for " + clearCount + " users");
        }
    }
    
    private int clearUsages(final List<NodeRef> personNodeRefs)
    {
        RetryingTransactionCallback<Integer> clearPersonUsage = new RetryingTransactionCallback<Integer>()
        {
            public Integer execute() throws Throwable
            {
                int clearCount = 0;
                for (NodeRef personNodeRef : personNodeRefs)
                {
                    nodeService.setProperty(personNodeRef, ContentModel.PROP_SIZE_CURRENT, null);
                    usageService.deleteDeltas(personNodeRef);
                    
                    if (logger.isTraceEnabled())
                    {
                        logger.trace("Cleared usage for person ("+ personNodeRef+")");
                    }
                    
                    clearCount++;
                }
                return clearCount;
            }
        };
        
        // execute in READ-WRITE txn
        return transactionService.getRetryingTransactionHelper().doInTransaction(clearPersonUsage, false);
    }
    
    /**
     * Recalculate content usage for all users that have no usage. 
     * Required if upgrading an existing Alfresco, for users that have not had their initial usage calculated.
     */
    private void calculateMissingUsages()
    {
        if (logger.isInfoEnabled()) 
        {
            logger.info("Enabled - calculate missing user usages ...");
        }
        
        final Map<String, NodeRef> users = new HashMap<String, NodeRef>();
        
        RetryingTransactionCallback<Object> getUsersWithoutUsage = new RetryingTransactionCallback<Object>()
        {
            public Object execute() throws Throwable
            {
                // get people (users) without calculated usage
                ObjectArrayQueryCallback userHandler = new ObjectArrayQueryCallback()
                {
                    public boolean handle(Object[] arr)
                    {
                        String username = (String)arr[0];
                        String uuid = (String)arr[1];
                        
                        users.put(username, new NodeRef(personStoreRef, uuid));
                        
                        return true; // continue to next node (more required)
                    }
                };
                
                nodeDaoService.getUsersWithoutUsage(tenantService.getName(personStoreRef), userHandler);
                        
                return null;
            }
        };
        
        // execute in READ-ONLY txn
        transactionService.getRetryingTransactionHelper().doInTransaction(getUsersWithoutUsage, true);
        
        if (logger.isInfoEnabled()) 
        {
            logger.info("Found " + users.size() + " users to recalculate");
        }
        
        int updateCount = 0;
        if (users.size() > 0)
        {
            updateCount = recalculateUsages(users);
        }
        
        if (logger.isInfoEnabled()) 
        {
            logger.info("... calculated missing usages for " + updateCount + " users");
        }
    }
    
    /*
     * Recalculate content usage for given users. Required if upgrading an existing Alfresco, for users that
     * have not had their initial usage calculated. In a future release, could also be called explicitly by
     * a SysAdmin, eg. via a JMX operation.
     */
    private int recalculateUsages(final Map<String, NodeRef> users)
    {
        final Map<String, Long> currentUserUsages = new HashMap<String, Long>(users.size());
        
        RetryingTransactionCallback<Long> calculateCurrentUsages = new RetryingTransactionCallback<Long>()
        {
            public Long execute() throws Throwable
            {
                List<String> stores = contentUsageImpl.getStores();
                 
                for (String store : stores)
                {
                    final StoreRef storeRef = tenantService.getName(new StoreRef(store));
                    
                    if (logger.isTraceEnabled())
                    {
                        logger.trace("Recalc usages for store=" + storeRef);
                    }
                    
                    // get content urls
                    ObjectArrayQueryCallback nodeContentUrlHandler = new ObjectArrayQueryCallback()
                    {
                        public boolean handle(Object[] arr)
                        {
                            String owner = (String)arr[0];
                            String creator = (String)arr[1];
                            String contentUrlStr = (String)arr[2];
                            
                            if (owner == null)
                            {
                                owner = creator;
                            }
                            
                            ContentData contentData = ContentData.createContentProperty(contentUrlStr);
                            if (contentData != null)
                            {
                                Long currentUsage = currentUserUsages.get(owner);
                                
                                if (currentUsage == null)
                                {
                                    currentUsage = 0L;
                                }
                                
                                currentUserUsages.put(owner, currentUsage + contentData.getSize());
                            }
                            return true; // continue to next node (more required)
                        }
                    };
                    nodeDaoService.getContentUrlsForStore(storeRef, nodeContentUrlHandler);
                }
                
                return null;
            }
        };
        
        // execute in READ-ONLY txn
        transactionService.getRetryingTransactionHelper().doInTransaction(calculateCurrentUsages, true);
        
        if (logger.isDebugEnabled()) 
        {
            logger.debug("Usages calculated - start update");
        }
        
        int updateCount = 0;
        int batchCount = 0;
        int totalCount = 0;
        
        List<Pair<NodeRef, Long>> batchUserUsages = new ArrayList<Pair<NodeRef, Long>>(updateBatchSize);
        
        for (Map.Entry<String, NodeRef> user : users.entrySet())
        {
            String userName = user.getKey();
            NodeRef personNodeRef = user.getValue();
            
            Long currentUsage = currentUserUsages.get(userName);
            if (currentUsage == null)
            {
                currentUsage = 0L;
            }
            
            batchUserUsages.add(new Pair<NodeRef, Long>(personNodeRef, currentUsage));
            batchCount++;
            totalCount++;
            
            if ((batchCount == updateBatchSize) || (totalCount == users.size()))
            {
                int updated = updateUsages(batchUserUsages);
                updateCount = updateCount + updated;
                
                batchUserUsages.clear();
                batchCount = 0;
            }
        }
        
        return totalCount;
    }
    
    private int updateUsages(final List<Pair<NodeRef, Long>> userUsages)
    {
        RetryingTransactionCallback<Integer> updateCurrentUsages = new RetryingTransactionCallback<Integer>()
        {
            public Integer execute() throws Throwable
            {
                int updateCount = 0;
                
                for (Pair<NodeRef, Long> userUsage : userUsages)
                {
                    NodeRef personNodeRef = userUsage.getFirst();
                    Long currentUsage = userUsage.getSecond();
                    
                    contentUsageImpl.setUserStoredUsage(personNodeRef, currentUsage);
                    usageService.deleteDeltas(personNodeRef);
                    
                    updateCount++;
                }
                return updateCount;
            }
        };
        
        // execute in READ-WRITE txn
        return transactionService.getRetryingTransactionHelper().doInTransaction(updateCurrentUsages, false);
    }
    
    /**
     * Collapse usages - note: for MT environment, will collapse all tenants
     */
    private void collapseUsages()
    {
        if (logger.isDebugEnabled()) 
        {
            logger.debug("Collapse usages ...");
        }
        
        // Collapse usage deltas (if a person has initial usage set)
        RetryingTransactionCallback<Set<NodeRef>> getUsageNodeRefs = new RetryingTransactionCallback<Set<NodeRef>>()
        {
            public Set<NodeRef> execute() throws Throwable
            {
                // Get distinct candidates
                return usageService.getUsageDeltaNodes();
            }
        };
        
        // execute in READ-ONLY txn
        Set<NodeRef> usageNodeRefs = transactionService.getRetryingTransactionHelper().doInTransaction(getUsageNodeRefs, true);
        
        int collapseCount = 0;
        for (final NodeRef usageNodeRef : usageNodeRefs)
        {
            Boolean collapsed = AuthenticationUtil.runAs(new RunAsWork<Boolean>()
            {
                public Boolean doWork() throws Exception
                {
                    return collapseUsage(usageNodeRef);
                }
            }, tenantAdminService.getDomainUser(AuthenticationUtil.getSystemUserName(), tenantAdminService.getDomain(usageNodeRef.getStoreRef().getIdentifier())));
            
            if (collapsed)
            {
                collapseCount++;
            }
        }
        
        if (logger.isDebugEnabled()) 
        {
            logger.debug("... collapsed usages for " + collapseCount + " users");
        }
    }
    
    private boolean collapseUsage(final NodeRef usageNodeRef)
    {
        RetryingTransactionCallback<Boolean> collapseUsages = new RetryingTransactionCallback<Boolean>()
        {
            public Boolean execute() throws Throwable
            {
                if (!nodeService.exists(usageNodeRef))
                {
                    // Ignore
                    return false;
                }
                QName nodeType = nodeService.getType(usageNodeRef);
                
                if (nodeType.equals(ContentModel.TYPE_PERSON))
                {
                    NodeRef personNodeRef = usageNodeRef;
                    String userName = (String)nodeService.getProperty(personNodeRef, ContentModel.PROP_USERNAME);
                    
                    long currentUsage = contentUsageImpl.getUserStoredUsage(personNodeRef);
                    if (currentUsage != -1)
                    {
                        // collapse the usage deltas
                        currentUsage = contentUsageImpl.getUserUsage(userName);
                        usageService.deleteDeltas(personNodeRef);
                        contentUsageImpl.setUserStoredUsage(personNodeRef, currentUsage);
                        
                        if (logger.isTraceEnabled()) 
                        {
                            logger.trace("Collapsed usage: username=" + userName + ", usage=" + currentUsage);
                        }
                    }
                    else
                    {
                        if (logger.isWarnEnabled())
                        {
                            logger.warn("Initial usage for user has not yet been calculated: " + userName);
                        }
                    }
                }
                return true;
            }
        };
        
        // execute in READ-WRITE txn
        return transactionService.getRetryingTransactionHelper().doInTransaction(collapseUsages, false);
    }
}