hdfs auditlog(審計日誌)

時間 2019-11-19

原文原文鏈接

hdfs審計日誌(Auditlog)記錄了用戶針對hdfs的全部操做，詳細信息包括操做成功與否、用戶名稱、客戶機地址、操做命令、操做的目錄等。對於用戶的每個操做，namenode都會將這些信息以key-value對的形式組織成固定格式的一條日誌，而後記錄到audit.log文件中。經過審計日誌，咱們能夠實時查看hdfs的各類操做情況、能夠追蹤各類誤操做、能夠作一些指標監控等等。java

hdfs的審計日誌功能是可插拔的，用戶能夠經過實現默認接口擴展出知足本身所需的插件來替換hdfs默認提供的審計日誌功能，或者與之並用。node

啓用審計日誌

若是僅僅只啓用默認的AuditLogger（DefaultAuditLogger），則在log4j.properties添加以下配置(hdfs.audit.logger必須配置爲INFO級別)便可，審計日誌會與namenode的系統日誌獨立開來保存，log4j.appender.RFAAUDIT.File可配置保存的位置及文件。 FSNamesystem根據log4j.properties中hdfs.audit.logger是否爲INFO，以及是否配置了DefaultAuditLogger以外的其餘AuditLogger，來決定是否啓用審計日誌功能。web

#
# hdfs audit logging
#
hdfs.audit.logger=INFO,NullAppender
hdfs.audit.log.maxfilesize=256MB
hdfs.audit.log.maxbackupindex=20
log4j.logger.org.apache.hadoop.hdfs.server.namenode.FSNamesystem.audit=${hdfs.audit.logger}
log4j.additivity.org.apache.hadoop.hdfs.server.namenode.FSNamesystem.audit=false
log4j.appender.RFAAUDIT=org.apache.log4j.RollingFileAppender
log4j.appender.RFAAUDIT.File=${hadoop.log.dir}/hdfs-audit.log
log4j.appender.RFAAUDIT.layout=org.apache.log4j.PatternLayout
log4j.appender.RFAAUDIT.layout.ConversionPattern=%d{ISO8601} %p %c{2}: %m%n
log4j.appender.RFAAUDIT.MaxFileSize=${hdfs.audit.log.maxfilesize}
log4j.appender.RFAAUDIT.MaxBackupIndex=${hdfs.audit.log.maxbackupindex}

審計日誌的接口及實現

Namenode開放了AuditLogger接口，並定義抽象類HdfsAuditLogger 實現AuditLogger，默認提供實現類DefaultAuditLogger。構造FSNamesystem時經過initAuditLoggers(Configuration conf)方法建立AuditLogger列表。在記錄用戶操做時，會將操做信息逐一傳給列表中的每個AuditLogger，由其作對應的審計處理。apache

經過實現Auditloger接口或者擴展HdfsAuditLogger類，用戶能夠實現本身的AuditLogger來知足所需，例若有針對性的記錄審計日誌（當集羣、訪問量上規模以後瘋狂刷日誌必然對namenode有影響，有針對性的記錄有必要的日誌是緩解此情況的一種可選方案）、擴展功能、將日誌接入實時系統作實時分析或監控等。用戶經過配置項dfs.namenode.audit.loggers在hdfs-site.xml中配置Auditloger的實現類，多個實現能夠經過逗號分開，更改配置後重啓namenode接口生效。FSNamesystem的initAuditLoggers(Configuration conf)方法經過該配置項加載並實例化實現類，初始化後存入集合。若是用戶沒有配置，那麼默認使用DefaultAuditLogger。若是啓動了nntop功能，還會使用TopAuditLogger。app

FSNamesystem 初始化全部的AuditLogger：ide

private List<AuditLogger> initAuditLoggers(Configuration conf) {
    // Initialize the custom access loggers if configured.
    //DFS_NAMENODE_AUDIT_LOGGERS_KEY=dfs.namenode.audit.loggers
    Collection<String> alClasses = conf.getStringCollection(DFS_NAMENODE_AUDIT_LOGGERS_KEY);
    List<AuditLogger> auditLoggers = Lists.newArrayList();
    if (alClasses != null && !alClasses.isEmpty()) {
      for (String className : alClasses) {
        try {
          AuditLogger logger;
          if (DFS_NAMENODE_DEFAULT_AUDIT_LOGGER_NAME.equals(className)) {
            logger = new DefaultAuditLogger();
          } else {
            logger = (AuditLogger) Class.forName(className).newInstance();
          }
          logger.initialize(conf);
          auditLoggers.add(logger);
        } catch (RuntimeException re) {
          throw re;
        } catch (Exception e) {
          throw new RuntimeException(e);
        }
      }
    }

    // Make sure there is at least one logger installed.
    // 若是用戶沒有提供AuditLoggers，則默認使用DefaultAuditLogger
    if (auditLoggers.isEmpty()) {
      auditLoggers.add(new DefaultAuditLogger());
    }

    // Add audit logger to calculate top users
    // 默認topConf.isEnabled是開啓的，用於指標聚合、上報
    // TopAuditLogger相似 top命令
    if (topConf.isEnabled) {
      topMetrics = new TopMetrics(conf, topConf.nntopReportingPeriodsMs);
      auditLoggers.add(new TopAuditLogger(topMetrics));
    }

    return Collections.unmodifiableList(auditLoggers);
  }

DefaultAuditLogger記錄日誌：oop

@Override
    public void logAuditEvent(boolean succeeded, String userName,
        InetAddress addr, String cmd, String src, String dst,
        FileStatus status, UserGroupInformation ugi,
        DelegationTokenSecretManager dtSecretManager) {
      if (auditLog.isInfoEnabled()) {
        final StringBuilder sb = auditBuffer.get();
        sb.setLength(0);
        sb.append("allowed=").append(succeeded).append("\t");
        sb.append("ugi=").append(userName).append("\t");
        sb.append("ip=").append(addr).append("\t");
        sb.append("cmd=").append(cmd).append("\t");
        sb.append("src=").append(src).append("\t");
        sb.append("dst=").append(dst).append("\t");
        if (null == status) {
          sb.append("perm=null");
        } else {
          sb.append("perm=");
          sb.append(status.getOwner()).append(":");
          sb.append(status.getGroup()).append(":");
          sb.append(status.getPermission());
        }
        if (logTokenTrackingId) {
          sb.append("\t").append("trackingId=");
          String trackingId = null;
          if (ugi != null && dtSecretManager != null
              && ugi.getAuthenticationMethod() == AuthenticationMethod.TOKEN) {
            for (TokenIdentifier tid: ugi.getTokenIdentifiers()) {
              if (tid instanceof DelegationTokenIdentifier) {
                DelegationTokenIdentifier dtid =
                    (DelegationTokenIdentifier)tid;
                trackingId = dtSecretManager.getTokenTrackingId(dtid);
                break;
              }
            }
          }
          sb.append(trackingId);
        }
        sb.append("\t").append("proto=");
        sb.append(NamenodeWebHdfsMethods.isWebHdfsInvocation() ? "webhdfs" : "rpc");
        logAuditMessage(sb.toString());
      }
    }

    public void logAuditMessage(String message) {
      auditLog.info(message);
    }

審計過程

客戶端對hdfs的全部操做，無論成功與否都會由FSNamesystem記錄下。以刪除操做爲例，FSNamesystem在正常刪除給定src後調用logAuditEvent(true, "delete", src)記錄這次成功的delete操做，若是刪除失敗拋出異常，則調用logAuditEvent(false, "delete", src)記錄這次失敗的delete操做。ui

boolean delete(String src, boolean recursive, boolean logRetryCache)
      throws IOException {
    waitForLoadingFSImage();
    BlocksMapUpdateInfo toRemovedBlocks = null;
    writeLock();
    boolean ret = false;
    try {
      checkOperation(OperationCategory.WRITE);
      checkNameNodeSafeMode("Cannot delete " + src);
      toRemovedBlocks = FSDirDeleteOp.delete(
          this, src, recursive, logRetryCache);
      ret = toRemovedBlocks != null;
    } catch (AccessControlException e) {
      logAuditEvent(false, "delete", src);
      throw e;
    } finally {
      writeUnlock();
    }
    getEditLog().logSync();
    if (toRemovedBlocks != null) {
      removeBlocks(toRemovedBlocks); // Incremental deletion of blocks
    }
    logAuditEvent(true, "delete", src);
    return ret;
  }
  


//判斷是不是外部調用，只對rpc調用和webHdfs調用作審計
  boolean isExternalInvocation() {
    return Server.isRpcInvocation() || NamenodeWebHdfsMethods.isWebHdfsInvocation();
  }  

  //判斷是否啓用審計日誌功能
  public boolean isAuditEnabled() {
    return !isDefaultAuditLogger || auditLog.isInfoEnabled();
  }
  
  //succeeded：操做是否成功     cmd:操做命令     src:操做對象
  private void logAuditEvent(boolean succeeded, String cmd, String src)
      throws IOException {
    logAuditEvent(succeeded, cmd, src, null, null);
  }
  
  private void logAuditEvent(boolean succeeded, String cmd, String src,
      String dst, HdfsFileStatus stat) throws IOException {
    if (isAuditEnabled() && isExternalInvocation()) {
      logAuditEvent(succeeded, getRemoteUser(), getRemoteIp(),
                    cmd, src, dst, stat);
    }
  }
  
  //獲取操做對象的信息，調用全部的auditloger 作審計
  private void logAuditEvent(boolean succeeded,
      UserGroupInformation ugi, InetAddress addr, String cmd, String src,
      String dst, HdfsFileStatus stat) {
    FileStatus status = null;
    if (stat != null) {
      Path symlink = stat.isSymlink() ? new Path(stat.getSymlink()) : null;
      Path path = dst != null ? new Path(dst) : new Path(src);
      status = new FileStatus(stat.getLen(), stat.isDir(),
          stat.getReplication(), stat.getBlockSize(), stat.getModificationTime(),
          stat.getAccessTime(), stat.getPermission(), stat.getOwner(),
          stat.getGroup(), symlink, path);
    }
    for (AuditLogger logger : auditLoggers) {
      if (logger instanceof HdfsAuditLogger) {
        HdfsAuditLogger hdfsLogger = (HdfsAuditLogger) logger;
        hdfsLogger.logAuditEvent(succeeded, ugi.toString(), addr, cmd, src, dst,
            status, ugi, dtSecretManager);
      } else {
        logger.logAuditEvent(succeeded, ugi.toString(), addr,
            cmd, src, dst, status);
      }
    }
  }

審計日誌接入實時系統的方法

方法1：擴展Log4J的appender，由appender將日誌發送到kafka。
方法2：直接讓kafka的producer讀取日誌文件。

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。