Merge pull request #978 from juanpablosalas/fi_tool

Moving FI tool scripts to CMSRucio project

Author: ericvaandering

.github/workflows/fileinvalidation-docker-image.yml

@@ -0,0 +1,34 @@
+name: File Invalidation Tool Docker Image CI
+
+on:
+  push:
+    tags:
+      - 'fileinvalidation-*'
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v3
+        
+      - name: Get tag name
+        id: tagname
+        run: echo "tag=${GITHUB_REF#refs/tags/loadtest-}" >> "$GITHUB_ENV"
+        
+      - name: Login to CERN Harbour
+        uses: docker/login-action@v2
+        with:
+          registry: registry.cern.ch
+          username: ${{ secrets.HARBOR_USERNAME }}
+          password: ${{ secrets.HARBOR_TOKEN }}
+
+      - name: Build the Docker Image
+        run: |
+          cd DMOps/file_invalidation_tool && docker build . \
+          --file Dockerfile \
+          --tag registry.cern.ch/${{ vars.HARBOR_REPOSITORY }}/file_invalidation_tool:${{ env.tag }}
+      
+      - name: Push Image to CERN Harbour
+        run: docker push registry.cern.ch/${{ vars.HARBOR_REPOSITORY }}/file_invalidation_tool:${{ env.tag }}
+

DMOps/file_invalidation_tool/Dockerfile

@@ -0,0 +1,39 @@
+FROM registry.cern.ch/cmsmonitoring/cmsmon-spark:latest
+
+# Set environment variables
+ENV PYCURL_SSL_LIBRARY=nss \
+    X509_USER_CERT=/certs/usercert.pem \
+    X509_USER_KEY=/certs/userkey.pem \
+    USER=dmtops \
+    RUCIO_CONFIG=/cvmfs/cms.cern.ch/rucio/rucio.cfg \
+    RUCIO_ACCOUNT=transfer_ops \
+    DRIVERPORT=5001 \
+    BMPORT=5002 \
+    UIPORT=5003
+
+ADD http://repository.egi.eu/sw/production/cas/1/current/repo-files/egi-trustanchors.repo /etc/yum.repos.d/egi.repo
+
+
+# Install dependencies for Rucio, DBS and Gfal
+RUN dnf install -y libcurl-devel openssl-devel libffi-devel ca-policy-egi-core\
+    && dnf install -y cmake gfal2-devel libcurl-devel \
+    && dnf install -y gfal2-all python3-gfal2 python3-gfal2-util \
+    && dnf -y groupinstall "Development Tools" || true \
+    && pip3 install cx-Oracle SQLAlchemy==1.4.49 dbs3-client rucio-clients \
+    && pip3 install --compile --global-option="--with-nss" --no-cache-dir pycurl
+
+# Expose ports
+EXPOSE 5001
+EXPOSE 5002
+EXPOSE 5003
+
+# Copy code
+COPY ./src /src/
+
+# Set working directory
+WORKDIR /src
+
+RUN chmod 755 /src/submit_invalidation.sh
+
+# Set entrypoint
+ENTRYPOINT ["python3", "run_invalidations.py"]

DMOps/file_invalidation_tool/README.md

@@ -0,0 +1,128 @@
+# File Invalidation Tool for Rucio and DBS
+## Overview
+
+This guide outlines the steps to run the file invalidation tool for Rucio and DBS using Docker image. The tool assists in invalidating specific files, datasets or containers within these systems, to ensure data consistency. Additionally, it has a running mode to check the integrity of files in a given RSE(checksum validation), and invalidate the corrupted replicas. Finally, the tool can also be used to invalidate all files in a given site.
+
+## Prerequisites, Folder Structure and tool input
+
+### Tool Input
+
+The tool has 5 running modes.  It's important that your cert and key (decrypted) have enough permissions to invalidate on DBS and declare replicas as corrupted on Rucio, additionally to this it they require the following inputs and parameters:
+
+| Running Mode | Description | Tool Mode | Input File | Params | Auth Requirements |
+| ----------- | ----------- | ----------- | ----------- | ----------- | ----------- |
+| Global Invalidation | Invalidate all files from received files, datasets or containers list on Rucio and DBS | `global` | `<filename>.txt`: txt file containing list of files, datasets or containers | `--reason <reason>`: comment for invalidation<br>`--dry-run`(**optional**): Simulate the execution without actually performing the file invalidation<br>`--erase-mode`(**optional**): Erase empty DIDs | `./certs/usercert.pem`<br>`./certs/userkey.pem`<br>`./secrets/dmtops.keytab`|
+| DBS Invalidation | Invalidate all files from received files, datasets or containers list only on DBS | `only-dbs` | `<filename>.txt`: txt file containing list of files, datasets or containers | `--reason <reason>`: comment for invalidation<br>`--dry-run`(**optional**): Simulate the execution without actually performing the file invalidation<br>`--erase-mode`(**optional**): Erase empty DIDs | `./certs/usercert.pem`<br>`./certs/userkey.pem`<br>`./secrets/dmtops.keytab`|
+| Rucio Invalidation | Invalidate all files from received files, datasets or containers list only on Rucio | `only-rucio` | `<filename>.txt`: txt file containing list of files, datasets or containers | `--reason <reason>`: comment for invalidation<br>`--dry-run`(**optional**): Simulate the execution without actually performing the file invalidation<br>`--erase-mode`(**optional**): Erase empty DIDs | `./certs/usercert.pem`<br>`./certs/userkey.pem`<br>`./secrets/dmtops.keytab`|
+| Integrity Validation | Validate integrity of files in the given RSE | `integrity-validation` | `<filename>.csv`: csv file containing list of files and RSE [FILENAME,RSE_EXPRESSION] | `--dry-run`(**optional**): Simulate the execution without actually performing the file invalidation in case of being corrupted | `./certs/usercert.pem`<br>`./certs/userkey.pem`|
+| Site Invalidation | Invalidate in Rucio all files from received list at a specific site | `site-invalidation` | `<filename>.txt`: txt file containing list of files, datasets or containers | `--rse <rse>`: RSE to invalidate at<br>`--reason <reason>`: comment for invalidation<br>`--dry-run`(**optional**): Simulate the execution without actually performing the file invalidation | `./certs/usercert.pem`<br>`./certs/userkey.pem`<br>`./secrets/dmtops.keytab`|
+
+> **Note:** The userkey.pem should be decrypted.
+
+??? Example
+    **USERKEY decryption**
+    `openssl rsa -in <encrypted_userkey> -out userkey.pem`
+
+    You would be asked to enter the password.
+
+??? Info
+    **Checksum Validation Mode**
+
+    Some files could be heavy and may lead to exceed your lxplus quota. In case of seeing this error move your working directory to `/eos/user/<first_username_letter>/<username>/` directory.
+    ```Bash
+    gfal-copy error: 122 (Disk quota exceeded) - errno reported by local system call Disk quota exceeded
+    ```
+
+### Environment
+
+This script is thought to be run on **lxplus** or CERN server with access to `registry.cern.ch` and `/cvmfs/` directory.
+
+Setting all together, the working directory structure can change a bit, but it should look like this:
+working_directory/
+├── dids.txt / replicas_validation.csv
+├── certs/
+│   ├── usercert.pem
+│   └── userkey.pem
+
+## Run File Invalidation tool
+
+### 1. CERN Registry Authentication
+
+1. Visit [cern registry](https://registry.cern.ch/).
+2. Login via OIDC Provider.
+3. Click on your username located in the top right.
+4. Click on **User Profile**
+5. Copy the **CLI Secret**, it will be used in the next step.
+
+### 2. Login into CERN Registry
+```Bash
+docker login registry.cern.ch -u <username>
+```
+- `docker login`: Logs in to the Docker registry.
+- `registry.cern.ch`: CERN registry URL.
+- `-u <username>`: CERN registry username.
+
+It will ask you to enter your password. **Enter your CLI Secret.**
+
+### 3. Run the container
+
+```Bash
+docker run -P \
+  -v "$(pwd)/<input_file>:/input/<input_file>" \
+  -v "$(pwd)/certs:/certs" \
+  [-v "$(pwd)/secrets:/secrets" \]
+  --mount type=bind,source=/cvmfs/,target=/cvmfs/,readonly \
+  --mount type=bind,source=/etc/gfal2.d/,target=/etc/gfal2.d/,readonly \
+  --mount type=bind,source=/etc/grid-security/certificates/,target=/etc/grid-security/certificates/,readonly \
+  --network host --rm registry.cern.ch/cmsrucio/file_invalidation_tool [Tool_Mode_Options]
+```
+- `docker run`: Executes a Docker container.
+- `-P`: Publishes all exposed ports to the host interfaces.
+- Volumes mounted:
+  - `-v "$(pwd)/<input_file>:/input/<input_file>"`: Mounts the containers_inv.txt file from the host to /input/dids.txt within the container.
+  - `-v "$(pwd)/certs:/certs"`: Mounts the certs directory from the host to /certs within the container. It must contain the usercert.pem and userkey.pem.
+  - `-v "$(pwd)/secrets:/secrets"`: Mounts the secrets directory from the host to /secrets within the container. It must contain the keytab file.
+  - `--mount type=bind,source=/cvmfs/,target=/cvmfs/,readonly`: Binds the /cvmfs/ directory on the host as read-only within the container.
+  - `--mount type=bind,source=/etc/gfal2.d/,target=/etc/gfal2.d/,readonly`: Binds the /etc/gfal2.d/ directory on the host as read-only within the container. Necessary for the integrity-validation mode.
+  - `--mount type=bind,source=/etc/grid-security/certificates/ ,target=/etc/grid-security/certificates/,readonly`: Binds the /etc/grid-security/certificates/ directory on the host as read-only within the container. Necessary for the proxy-init command.
+- `--network host`: Uses the host's network stack within the container.
+- `--rm`: Automatically removes the container when it exits.
+- `registry.cern.ch/cmsrucio/file_invalidation_tool`: Name of the Docker image to run.
+
+??? Example
+
+    ```Bash
+    docker run -P \
+      -v "$(pwd)/<input_file>:/input/<input_file>.txt" \
+      -v "$(pwd)/certs:/certs" \
+      -v "$(pwd)/secrets:/secrets" \
+      --mount type=bind,source=/cvmfs/,target=/cvmfs/,readonly \
+      --mount type=bind,source=/etc/grid-security/certificates/,target=/etc/grid-security/certificates/,readonly \
+      --network host --rm registry.cern.ch/cmsrucio/file_invalidation_tool [global | only-dbs | only-rucio] --reason <reason>
+    ```
+
+    ```Bash
+    docker run -P \
+      -v "$(pwd)/<input_file>:/input/<input_file>.csv" \
+      -v "$(pwd)/certs:/certs" \
+      --mount type=bind,source=/cvmfs/,target=/cvmfs/,readonly \
+      --mount type=bind,source=/etc/gfal2.d/,target=/etc/gfal2.d/,readonly \
+      --mount type=bind,source=/etc/grid-security/certificates/,target=/etc/grid-security/certificates/,readonly \
+      --network host --rm registry.cern.ch/cmsrucio/file_invalidation_tool integrity-validation
+    ```
+
+    ```Bash
+    docker run -P \
+      -v "$(pwd)/<input_file>:/input/<input_file>.txt" \
+      -v "$(pwd)/certs:/certs" \
+      -v "$(pwd)/secrets:/secrets" \
+      --mount type=bind,source=/cvmfs/,target=/cvmfs/,readonly \
+      --mount type=bind,source=/etc/grid-security/certificates/,target=/etc/grid-security/certificates/,readonly \
+      --network host --rm registry.cern.ch/cmsrucio/file_invalidation_tool site-invalidation --rse <rse>  --reason <reason>
+    ```
+## Additional Notes
+
+- The tool's output will provide details about the invalidation process.
+- User Authorization: Ensure you have the necessary permissions to invalidate on DBS.
+  - The provided certificates will be used for DBS invalidation, in case of authorization errors, rucio invalidation will not be executed.
+  - Rucio Invalidation will be done using the  the dmtops certificate and transfer_ops account since many users will not have permissions to develop this operation.

DMOps/file_invalidation_tool/src/container_invalidation.py

@@ -0,0 +1,124 @@
+import click as click
+from rucio.client import Client
+import pandas as pd
+import numpy as np
+from pyspark.sql.functions import col, collect_list, concat_ws
+from CMSSpark.spark_utils import get_spark_session
+from hadoop_queries import get_df_rse_locks, get_df_rse_replicas, get_df_contents
+from pyspark.sql.window import Window
+
+
+@click.command()
+@click.option('--filename', required=True, default=None, type=str,
+              help='Name of the text file having the datasets names')
+@click.option('--rse', required=False, default=None, type=str,
+              help='RSE to look at')
+@click.option('--mode', required=False, type=click.Choice(['rucio','spark']), default='rucio', help='List generation mode')
+def invalidate_containers(filename,rse, mode):
+    #TODO: Check rse option
+
+    if mode=='rucio':
+        #Start Rucio Client
+        rucio_client = Client()
+        
+        # Read the containers to delete
+        with open('/input/'+filename, 'r') as file:
+            containers_to_delete = file.readlines()
+        containers_to_delete = list(map(str.strip,containers_to_delete))
+
+        # Get the list of containers
+        dict_delete = [{'scope':'cms','name':lfn} for lfn in containers_to_delete]
+
+        #Get the content of the containers to delete (content includes filename, dataset and container)
+        container_files = list(rucio_client.bulk_list_files(dict_delete))
+
+        #Files to invalidate on DBS
+        df_container_files = pd.DataFrame(columns=['name','parent_name'])
+        df_container_files = pd.concat([df_container_files,pd.DataFrame(container_files)])[['name','parent_name']].rename(columns={'parent_name':'CONTAINER','name':'FILENAME'})
+        df_container_files['FILENAME'].drop_duplicates().to_csv('/input/dbs_files_inv.txt',index=False, header = False)
+
+        #Replicas to declare as bad
+        file_list = [{'scope':'cms','name':name} for name in df_container_files['FILENAME']]
+        df_replicas = pd.DataFrame(columns=['name','states'])
+        if len(file_list)>0:
+            for arr in np.array_split(file_list,indices_or_sections=int(len(file_list)/1000)+1):
+                list_replicas_i = list(rucio_client.list_replicas(dids=arr.tolist(),all_states=True))
+                df_replicas_i = pd.DataFrame(list_replicas_i)[['name','states']]
+                df_replicas = pd.concat([df_replicas,df_replicas_i])
+        
+        df_replicas['rses']=df_replicas['states'].apply(dict.keys)
+        df_replicas['rses']=df_replicas['rses'].apply(lambda s: ';'.join(s))
+
+        if rse is not None:
+            df_replicas = df_replicas[df_replicas.rses.str.contains(rse)]
+            df_replicas.loc[:,'rses'] = rse
+
+        df_replicas[['name','rses']].rename(columns={'name':'FILENAME','rses':'RSES'}).drop_duplicates().to_csv('/input/rucio_replicas_inv.csv',index=False)
+
+        #Datasets to erase from Rucio
+        info_dataset = rucio_client.get_locks_for_dids(dict_delete)
+        if len(info_dataset)>0:
+            df_datasets = pd.DataFrame(info_dataset)
+            if rse is not None:
+                df_datasets = df_datasets[df_datasets['rse']==rse]
+        else:
+            df_datasets = pd.DataFrame(columns=['name'])
+        df_datasets[['name']].rename(columns={'name':'DATASET'}).drop_duplicates().to_csv('/input/datasets_inv.txt',index=False,header=False)
+
+        #Rules to erase
+        #RSE is exported in case it's tape and require purge_replicas
+        dataset_list = [{'scope':'cms','name':n,'type':'dataset'} for n in df_datasets['name'].drop_duplicates().values]
+        df_rules = pd.DataFrame(columns=['rse','rule_id'])
+        if len(dataset_list)>0:
+            for arr in np.array_split(dataset_list,indices_or_sections=int(len(dataset_list)/400)+1):
+                info_rules_i = rucio_client.get_locks_for_dids(arr.tolist()) 
+                if len(info_rules_i)>0:
+                    df_rules_i = pd.DataFrame(info_rules_i)[['rse','rule_id']]
+                    df_rules = pd.concat([df_rules,df_rules_i])
+
+        if rse is not None:
+            df_rules['includes_rse'] = df_rules['rse_expression'].apply(lambda exp: {'rse':rse} in list(rucio_client.list_rses(rse_expression=exp)))
+            df_rules = df_rules.loc[df_rules.includes_rse]
+
+        df_rules.columns = df_rules.columns.str.upper()
+        df_rules[['RULE_ID','RSE']].drop_duplicates().to_csv('/input/rucio_rules_delete.csv',index=False)
+    else:
+        spark = get_spark_session(app_name='global_containers_invalidation')
+
+        #Read the containers to delete
+        filename = f'/user/dmtops/{filename}'
+        df_delete = spark.read.text(filename)
+        df_delete = df_delete.withColumnRenamed('value','CONTAINER')
+
+        #Get the basic df
+        df_locks = get_df_rse_locks(spark)
+        df_replicas = get_df_rse_replicas(spark,rse)
+        df_contents = get_df_contents(spark).alias('co')
+
+        #Get the content of the containers to delete (content includes filename, dataset and container)
+        df_delete = df_delete.join(df_contents,df_delete.CONTAINER==df_contents.CONTAINER,how='inner').select(['co.*']).alias('de')
+
+        #Replicas to declare as bad
+        df_delete = df_delete.join(df_replicas,df_delete.FILENAME==df_replicas.NAME,how='inner').select(['de.*','RSE','REPLICA_STATE']).alias('de')
+
+        #Rules protecting the replicas
+        df_delete = df_delete.join(df_locks,(df_delete.FILENAME==df_locks.NAME) & (df_delete.RSE == df_locks.RSE),how='left').select(['de.*','RULE_ID']).alias('de')
+        df_delete.cache()
+
+        #Files to invalidate on DBS
+        df_delete.select('FILENAME').distinct().toPandas().to_csv('/input/dbs_files_inv.txt',index=False, header = False)
+
+        windowSpec = Window.partitionBy('FILENAME')
+        df_delete.withColumn("RSES", collect_list(col("RSE")).over(windowSpec)) \
+            .select(['FILENAME','RSES']).withColumn("RSES", concat_ws(";", "RSES")).distinct().toPandas().to_csv('/input/rucio_replicas_inv.csv',index=False)
+
+        #Replicas to erase from Rucio
+        df_delete.select('DATASET').distinct().toPandas().to_csv('/input/datasets_inv.txt',index=False,header=False)
+
+        #RSE is exported in case it's tape and require purge_replicas
+        df_delete.filter(col('RULE_ID').isNotNull()).select(['RULE_ID','RSE']).distinct()\
+            .toPandas().to_csv('/input/rucio_rules_delete.csv',index=False)
+
+
+if __name__ == "__main__":
+    invalidate_containers()