How to capture CNI plugin's input and output

Summary

When using the CNI plugin in k8s, we may want to check the input and output of the CNI plugin. While there are several ways to do this, I will explain the case of using plugin chaining.

Plugin chaining

CNI has a mechanism called plugin chaining, which allows you to run multiple plugins in sequence. For example, lining up cni_a, cni_b, and cni_c will execute these in order. If CNI_COMMAND is ADD, it will execute cni_a, cni_b, cni_c in that order, and if DEL, it will execute them in the reverse order.
Using this function, you can check the input and output of Calico by arranging them in the order of cni_pre, Calico, and cni_post, for example.

Preparation

These are examples of python implementation of cni_pre and cni_post.

cni_pre.py

#!/usr/bin/python3
import os
import subprocess
import json
import sys
import tempfile
import netifaces
import hashlib
import random
import fcntl

def parseArgs():
	args = {}

	args['CNI_COMMAND'] = os.environ.get('CNI_COMMAND')
	if args['CNI_COMMAND'] == None:
		with open(LOG_FILE_NAME, mode='a') as f:
			f.write('env CNI_COMMAND not found')
		exit(1)

	args['CNI_CONTAINERID'] = os.environ.get('CNI_CONTAINERID')
	if args['CNI_CONTAINERID'] == None:
		with open(LOG_FILE_NAME, mode='a') as f:
			f.write('env CNI_CONTAINERID not found')
		exit(1)

	args['STDIN_DATA'] = json.load(sys.stdin)

	return args

def cmdAdd(args):
	with open(LOG_FILE_NAME, mode='a') as f:
		f.write('CNI_COMMAND=')
		f.write(args['CNI_COMMAND'])
		f.write(',CNI_CONTAINERID=')
		f.write(args['CNI_CONTAINERID'])
		f.write(',input=')
		f.write(json.dumps(args['STDIN_DATA']))
		f.write('\n')
	print(json.dumps({'cniVersion': '0.4.0'}))

def cmdDel(args):
	print("{}")

def cmdCheck(args):
	print("{}")

LOG_FILE_NAME = '/tmp/cni.log'
LOCK_FILE_NAME = '/var/run/lock/cni.lock'

args = parseArgs()

if args['CNI_COMMAND'] == 'VERSION':
	print('{"cniVersion": "0.4.0", "supportedVersions": ["0.1.0", "0.2.0", "0.3.0", "0.3.1", "0.4.0"]}')
	exit(0)

lockfp = open(LOCK_FILE_NAME, "w")
fcntl.flock(lockfp.fileno(), fcntl.LOCK_EX)

if args['CNI_COMMAND'] == 'ADD':
	cmdAdd(args)
elif args['CNI_COMMAND'] == 'DEL':
	cmdDel(args)
elif args['CNI_COMMAND'] == 'CHECK':
	cmdCheck(args)

fcntl.flock(lockfp.fileno(), fcntl.LOCK_UN)
lockfp.close()

cni_post.py

#!/usr/bin/python3
import os
import subprocess
import json
import sys
import tempfile
import netifaces
import hashlib
import random
import fcntl

def parseArgs():
	args = {}

	args['CNI_COMMAND'] = os.environ.get('CNI_COMMAND')
	if args['CNI_COMMAND'] == None:
		with open(LOG_FILE_NAME, mode='a') as f:
			f.write('env CNI_COMMAND not found')
		exit(1)

	args['CNI_CONTAINERID'] = os.environ.get('CNI_CONTAINERID')
	if args['CNI_CONTAINERID'] == None:
		with open(LOG_FILE_NAME, mode='a') as f:
			f.write('env CNI_CONTAINERID not found')
		exit(1)

	args['STDIN_DATA'] = json.load(sys.stdin)

	return args

def cmdAdd(args):
	with open(LOG_FILE_NAME, mode='a') as f:
		f.write('CNI_COMMAND=')
		f.write(args['CNI_COMMAND'])
		f.write(',CNI_CONTAINERID=')
		f.write(args['CNI_CONTAINERID'])
		f.write(',output=')
		f.write(json.dumps(args['STDIN_DATA']['prevResult']))
		f.write('\n')
	print(json.dumps(args['STDIN_DATA']['prevResult']))

def cmdDel(args):
	with open(LOG_FILE_NAME, mode='a') as f:
		f.write('CNI_COMMAND=')
		f.write(args['CNI_COMMAND'])
		f.write(',CNI_CONTAINERID=')
		f.write(args['CNI_CONTAINERID'])
		f.write(',input=')
		f.write(json.dumps(args['STDIN_DATA']))
		f.write('\n')
	print("{}")

def cmdCheck(args):
	print("{}")

LOG_FILE_NAME = '/tmp/cni.log'
LOCK_FILE_NAME = '/var/run/lock/cni.lock'

args = parseArgs()

if args['CNI_COMMAND'] == 'VERSION':
	print('{"cniVersion": "0.4.0", "supportedVersions": ["0.1.0", "0.2.0", "0.3.0", "0.3.1", "0.4.0"]}')
	exit(0)

lockfp = open(LOCK_FILE_NAME, "w")
fcntl.flock(lockfp.fileno(), fcntl.LOCK_EX)

if args['CNI_COMMAND'] == 'ADD':
	cmdAdd(args)
elif args['CNI_COMMAND'] == 'DEL':
	cmdDel(args)
elif args['CNI_COMMAND'] == 'CHECK':
	cmdCheck(args)

fcntl.flock(lockfp.fileno(), fcntl.LOCK_UN)
lockfp.close()

In order to run these before and after Calico, edit the Calico configuration file.
Add cni_pre.py to the top and cni_post.py to the bottom of the plugins array in /etc/cni/net.d/10-calico.conflist.

--- 10-calico.conflist.prev	2021-05-07 17:57:21.252206104 +0900
+++ 10-calico.conflist	2021-05-07 17:56:43.820039515 +0900
@@ -3,6 +3,9 @@
   "cniVersion": "0.3.1",
   "plugins": [
     {
+      "type": "cni_pre.py"
+    },
+    {
       "type": "calico",
       "log_level": "info",
       "log_file_path": "/var/log/calico/cni/cni.log",
@@ -27,6 +30,9 @@
     {
       "type": "bandwidth",
       "capabilities": {"bandwidth": true}
+    },
+    {
+      "type": "cni_post.py"
     }
   ]
 }

Submit

Create a pod with this command.

$ kubectl run nginx --image=nginx

Open the /tmp/cni.log on the specific node.
We can see Calico receives {"cniVersion": "0.3.1", "name": "k8s-pod-network", "type": "cni_pre.py"} and returns {"cniVersion": "0.3.1", "interfaces": [{"name": "calic440f455693"}], "ips": [{"version": "4", "address": "10.244.19.140/32"}], "dns": {}} when it processes the ADD command.

CNI_COMMAND=ADD,CNI_CONTAINERID=280718f61fd143ce1e75e9a40557129559098a9fb911d909a07f84fd6f8532e2,input={"cniVersion": "0.3.1", "name": "k8s-pod-network", "type": "cni_pre.py"}
CNI_COMMAND=ADD,CNI_CONTAINERID=280718f61fd143ce1e75e9a40557129559098a9fb911d909a07f84fd6f8532e2,output={"cniVersion": "0.3.1", "interfaces": [{"name": "calic440f455693"}], "ips": [{"version": "4", "address": "10.244.19.140/32"}], "dns": {}}