src/main/java/org/apache/hadoop/chukwa/datastore/ViewStore.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.hadoop.chukwa.datastore;

import java.io.File;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.LinkedHashMap;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
import org.apache.hadoop.chukwa.hicc.HiccWebServer;
import org.apache.hadoop.chukwa.rest.bean.ViewBean;
import org.apache.hadoop.chukwa.util.ExceptionUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.json.simple.JSONArray;

public class ViewStore {
  private String vid = null;
  private String uid = null;
  private ViewBean view = null;
  private static Log log = LogFactory.getLog(ViewStore.class);
  private static Configuration config = null;
  private static ChukwaConfiguration chukwaConf = new ChukwaConfiguration();
  private static String viewPath = null;
  private static String publicViewPath = viewPath+File.separator+"public";
  private static String usersViewPath = viewPath+File.separator+"users";
  private static String PUBLIC = "public".intern();

  static {
    config = HiccWebServer.getConfig();
    viewPath = config.get("fs.defaultFS")+File.separator+chukwaConf.get("chukwa.data.dir")+File.separator+"hicc"+File.separator+"views";
  }
  public ViewStore() throws IllegalAccessException {
  }

  public ViewStore(String uid, String vid) throws IllegalAccessException {
    this.uid=uid;
    this.vid=vid;
    load(uid, vid);
  }

  public void load(String uid, String vid) throws IllegalAccessException {
    StringBuilder vp = new StringBuilder();
    vp.append(usersViewPath);
    vp.append(File.separator);
    vp.append(uid);
    vp.append(File.separator);
    vp.append(vid);
    vp.append(".view");
    Path viewFile = new Path(vp.toString());
    try {
      FileSystem fs = FileSystem.get(config);
      if(!fs.exists(viewFile)) {
        StringBuilder pubPath = new StringBuilder();
        pubPath.append(publicViewPath);
        pubPath.append(File.separator);
        pubPath.append(vid);
        pubPath.append(".view");
        viewFile = new Path(pubPath.toString());        
      }
      if(fs.exists(viewFile)) {
        FileStatus[] fstatus = fs.listStatus(viewFile);
        long size = fstatus[0].getLen();
        FSDataInputStream viewStream = fs.open(viewFile);
        byte[] buffer = new byte[(int)size];
        viewStream.readFully(buffer);
        viewStream.close();
        try {
          view = new ViewBean(buffer);
          view.update();
        } catch (Exception e) {
          log.error(ExceptionUtil.getStackTrace(e));
          throw new IllegalAccessException("Unable to access view: "+vid);
        }
      }
    } catch (IOException ex) {
      log.error(ExceptionUtil.getStackTrace(ex));
    }
  }
  
  public String getMessage() {
    return view.toString();  
  }
  
  public ViewBean get() throws IllegalAccessException {
    if(view==null) {
      load(uid, vid);
    }
    if(view==null) {
      // Display global default view if user default view does not exist.
      try {
        load(null, vid);
      } catch(Exception e) {
        log.error(ExceptionUtil.getStackTrace(e));
        view = null;
      }
    }
    return view;
  }
  
  public void set(ViewBean view) throws IllegalAccessException {
    try {
      if (this.view == null || (this.view.getOwner().intern() == view.getOwner().intern())) {
        if(this.view!=null) {
          delete();
        }
        StringBuilder viewPath = new StringBuilder();
        if(view.getPermissionType().intern()==PUBLIC) {
          viewPath.append(publicViewPath);
        } else {
          viewPath.append(usersViewPath);
          viewPath.append(File.separator);
          viewPath.append(uid);
        }
        viewPath.append(File.separator);
        viewPath.append(view.getName());
        viewPath.append(".view");
        Path viewFile = new Path(viewPath.toString());
        try {
          FileSystem fs = FileSystem.get(config);
          FSDataOutputStream out = fs.create(viewFile,true);
          out.write(view.deserialize().toString().getBytes(Charset.forName("UTF-8")));
          out.close();
        } catch (IOException ex) {
          log.error(ExceptionUtil.getStackTrace(ex));
        }
        this.view = view;
      } else {
        if(view.getPermissionType().intern()==PUBLIC) {
          throw new IllegalAccessException("Unable to save public view, duplicated view exists.");
        } else {
          throw new IllegalAccessException("Unable to save user view.");          
        }
      }
    } catch (Exception e) {
      log.error(ExceptionUtil.getStackTrace(e));
      throw new IllegalAccessException("Unable to access user view.");
    }
  }

  public void delete() throws IllegalAccessException {
    try {
      if(this.view==null) {
        get();
      }
      if (this.view!=null) {
        StringBuilder viewPath = new StringBuilder();
        if(view.getPermissionType().intern()==PUBLIC) {
          viewPath.append(publicViewPath);
        } else {
          viewPath.append(usersViewPath);
          viewPath.append(File.separator);
          viewPath.append(uid);
        }
        viewPath.append(File.separator);
        viewPath.append(view.getName());
        viewPath.append(".view");
        Path viewFile = new Path(viewPath.toString());
        try {
          FileSystem fs = FileSystem.get(config);
          fs.delete(viewFile, true);
        } catch (IOException ex) {
          log.error(ExceptionUtil.getStackTrace(ex));
        }
      } else {
        throw new IllegalAccessException("Unable to delete user view, view does not exist.");          
      }
    } catch (Exception e) {
      log.error(ExceptionUtil.getStackTrace(e));
      throw new IllegalAccessException("Unable to access user view.");
    }
  }
  
  public static JSONArray list(String uid) throws IllegalAccessException {
    StringBuilder viewPath = new StringBuilder();
    viewPath.append(usersViewPath);
    viewPath.append(File.separator);
    viewPath.append(uid);
    String[] pathList = new String[2];
    pathList[0]=viewPath.toString();
    pathList[1]=publicViewPath;
    JSONArray list = new JSONArray();
    for(String path : pathList) {
      Path viewFile = new Path(path);
      try {
        FileSystem fs = FileSystem.get(config);
        FileStatus[] fstatus = fs.listStatus(viewFile);
        if(fstatus!=null) {
          for(int i=0;i<fstatus.length;i++) {
            if(!fstatus[i].getPath().getName().endsWith(".view")) {
              continue;
            }
            long size = fstatus[i].getLen();
            FSDataInputStream viewStream = fs.open(fstatus[i].getPath());
            byte[] buffer = new byte[(int)size];
            viewStream.readFully(buffer);
            viewStream.close();
            try {
              ViewBean view = new ViewBean(buffer);
              Map<String, String> json=new LinkedHashMap<String, String>();
              json.put("name", view.getName());
              json.put("type", view.getPermissionType());
              json.put("owner", view.getOwner());
              if(uid.intern()==view.getOwner().intern()) {
                json.put("editable","true");
              } else {
                json.put("editable","false");
              }
              list.add(json);
            } catch (Exception e) {
              log.error(ExceptionUtil.getStackTrace(e));
            }
          }
        }
      } catch (IOException ex) {
        log.error(ExceptionUtil.getStackTrace(ex));
        throw new IllegalAccessException("Unable to access user view."); 
      }
    }
    return list;
  }
}