I was getting pretty bad performance using Ubuntu 20.04 with a Dell XPS 9343. I upgraded to the yet-to-be-released Ubuntu 20.10 hoping it will help.
It did - performance is much better now.

This short post explains how to upgrade Ubuntu 20.04 focal to Ubuntu 20.10 groovy.

Steps

1. Backup your important data to the cloud or external drive. Upgrading is always a risky process, better be safe than sorry.
2. Update package information
   sudo apt update
3. Make sure all packages are up-to-date
   sudo apt upgrade
4. Replace apt repositories list from focal to groovy
sudo sed -i 's/focal/groovy/g' /etc/apt/sources.list
5. Update packages information from the new groovy repos
   sudo apt update
6. Upgrade to Ubuntu 20.10
   sudo apt dist-upgrade
7. Validate Ubuntu version

# lsb_release -ca
No LSB modules are available.
Distributor ID:	Ubuntu
Description:	Ubuntu Groovy Gorilla (development branch)
Release:	20.10
Codename:	groovy

I'm running Ubuntu 20.10 over a week and so far so good.
Let me know how it works out for you!

Troubleshooting

• Broadcom Wi-Fi driver doesn't work
  I had a minor problem with the Broadcom Wi-Fi driver crashing and not working after the upgrade.
  To solve that:

sudo apt-get remove --purge bcmwl-kernel-source
sudo apt-get install broadcom-sta-source
sudo apt-get install broadcom-sta-dkms
sudo apt-get install broadcom-sta-common
# reboot

While working on my little side project JSONPerf (check it out: jsonperf.com), I stumbled upon an interesting problem I thought was worth sharing with you.

A quick summary of what JSONPerf is will help us understand the need before discussing the solution.
The project's goal is to benchmark JSON libraries' performance in different programming languages. One of the features it provides is uploading a JSON file and benchmarking the different libraries on the user's specific file.

To implement this feature, I decided to implement the per-language benchmark endpoint using AWS Lambda as it provides me flexibility and I can use a different runtime for each function.

The problem I stumbled upon is the fact that in my serverless framework definitions file, I need to have multiple endpoints ( /python3, /python2, /java etc.) under the same API, each using a different language (runtime).

Unfortunately, this quickly turned out to be impossible because the python requirements plugin currently doesn't support my setup.
It allows you to specify different requirements file per function, but only if the file is called requirements.txt and the code is in different folders (github issue).
In my case, both Python 2 and Python 3 functions share the same code but have different requirements (different libraries to compare).

To solve it, I created a main serverless.yml that declares an API Gateway which is then shared between the other, per-function serverless.yml files.

Main (API) serverless.yml file

service: api 

provider:
name: aws
stage: dev
region: us-east-1

resources:
Resources:
  ApiGw:
    Type: AWS::ApiGateway::RestApi
    Properties:
      Name: ApiGw

Outputs:
  apiGatewayRestApiId:
    Value:
      Ref: ApiGw
    Export:
      Name: ApiGw-restApiId

  apiGatewayRestApiRootResourceId:
    Value:
      Fn::GetAtt:
        - ApiGw
        - RootResourceId
    Export:
      Name: ApiGw-rootResourceId

Services files, Python 2 example python/serverless_py2.yml

service: python2

plugins:
  - serverless-python-requirements
  - serverless-wsgi

package:
  exclude:
    - 'venv*/**'

custom:
  wsgi:
    app: app.app
    packRequirements: false
  pythonRequirements:
    fileName: requirements_py2.txt
    dockerizePip: non-linux
    slim: true

provider:
  name: aws
  stage: dev
  region: us-east-1
  runtime: python2.7
  apiGateway:
    restApiId:
      'Fn::ImportValue': ApiGw-restApiId
    restApiRootResourceId:
      'Fn::ImportValue': ApiGw-rootResourceId
    websocketApiId:
      'Fn::ImportValue': ApiGw-websocketApiId

functions:
  python2:
    handler: wsgi_handler.handler
    events:
      - http:
          method: post
          path: /python2
          cors: false

This is of course just an example and you can tweak it as much as you want with the flexibility you gain.

Hope this will help someone in finding the little hidden documentation to accomplish that using serverless framework.

In my previous post, I explained a bit about Flink and the difference between a job and session clusters. In addition, I showed how to deploy a job cluster in a way that works best in my opinion.

In this blog post, I will talk about how to achieve high availability on Kubernetes without using ZooKeeper by utilizing a custom, file-based high availability implementation. You can find the implementation here.

Why Would One Want to Run Flink Without ZooKeeper?

When running Flink on Kubernetes I think we should strive to use the powers Kubernetes gives us. One of them is ReplicaSet, which gives us the ability to deploy a pod with specified replicas and keep this number of pods up, even if a node fails.

Flink uses ZooKeeper to support job manager(s) high availability. In case a job manager fails, a new one can be started and become the leader.
With Kubernetes pod scheduling, we don't need ZooKeeper to manage job manager high availability.

By using StatefulSet for the job manager and Deployment for the task managers, we make use of ReplicaSet behind the scenes, hence making sure our managers will stay up even without ZooKeeper.

How to Implement a Custom High Availability Service?

To implement our custom HA service we need to implement a few things:

Leader Retrievers and Election Services

This tells flink how to elect a job manager to be the leader and where to retrieve the leader from.

In our case, when we have one job manager and he's always the leader we just tell flink to always choose the same leader, without election. This is why I used StandaloneLeaderElectionService for all the election services.
From the documentation: "The standalone implementation assumes that there is only a single LeaderContender and thus directly grants him the leadership upon start up".

For leader retrievers, I used StandaloneLeaderRetrievalService with the relevant constant address we can use thanks to the Kubernetes services we deploy.
From the documentation: "This implementation assumes that there is only a single contender for leadership (e.g., a single JobManager or ResourceManager process) and that this process is reachable under a constant address."

Checkpoint Recovery Factory

This is a factory that provides 2 classes that allow Flink to save checkpoints, retrieve checkpoints and keep a count of checkpoints.

In our case, both the checkpoints and the counter are saved on disk. You can find the implementations in FsCheckpointRecoveryFactory.java and FsCheckpointIDCounter.java.

Submitted Job Graph Store

This provides a way to save the job graphs and retrieve them.
As we are running a job cluster, we have only one job. This means we can always provide the same job graph.

Flink has a class that helps us with that, SingleJobSubmittedJobGraphStore. You can find the implementation here.

Running Jobs Registry

Documentation explains the need for that component quite well: "This registry is used in highly-available setups with multiple master nodes, to determine whether a new leader should attempt to recover a certain job (because the job is still running), or whether the job has already finished successfully (in case of a finite job) and the leader has only been granted leadership because the previous leader quit cleanly after the job was finished."

I used the provided FsNegativeRunningJobsRegistry class.
Its documentation can be found here and implementation here.

Kubernetes Setup

To make all of this work, we need to make a few adjustments in our k8s YAMLs.

• Change our Deployment to StatefulSet. This tells Kubernetes that the pods are stateful and the we have a persistent volume attached to each one of them (currently one).
• We want to make sure our shared volume is accessible by the user Flink runs with. To achieve that I added an init container that changes the ownership of that directory.
• We add a StorageClass and a PersistentVolume to be able to mount a volume to our pods. We mount the shared volume at /flink-shared.

Flink Setup

A few things we need to change in our flink configuration to utilize our brand new HA service:

• Change our high-availability to our factory class
  high-availability: com.ronlut.flinkjobcluster.filesystemha.SingleFsHaServicesFactory
• Set the HA storage dir to our mounted persistent volume
  high-availability.storageDir: file:///flink-shared/ha
• Set up the backend state storage to filesystem
  state.backend: filesystem
• Set the checkpoints and savepoints dir to the a shared directory that is mounted to the job and task managers
  state.checkpoints.dir: file:///flink-shared/checkpoints and state.savepoints.dir: file:///flink-shared/savepoints

Complete config file can be found here.

Final Notes

• In this example I used aws-ebs in the StorageClass to show how this will work in a cloud environment. Change it to the equivalent provisioner for your cloud provider
• If you are using EBS, job manager and task managers must be on same node as EBS can't be mounted to more than one node.
  This is achieved by setting affinity on the task manager deployment pods.
  This might not be needed in other cloud providers, depending on your storage.

Full working example can be found here.

Let me know if you have any questions or something is missing.

I love Flink. I think it's an amazing product, with great documentation and community.

For readers who aren't familiar with Flink, it is a framework for computations over unbounded and bounded data streams. It runs in a distributed manner and designed to perform exceptionally at scale.
You can read more about Flink here.

I didn't think I would struggle with doing something pretty straightforward like deploying a job cluster on k8s, not to mention deploying it on k8s with file based high-availability configured, which will be covered in the next post.

TL;DR: GitHub repo

Just to be on the same page, let's explain what a job cluster is and how is it different from a session cluster.

Job VS Session Cluster

Session cluster is a long-running Flink cluster, executing the jobs submitted to it.
Job cluster on the other hand, is a Flink cluster that is dedicated to run a single predefined job, without job submission.

Why would you choose one over the other?

In my opinion, a session cluster is more suitable to a situation where you submit multiple short-running jobs, dealing with bounded data. The cluster's resources are shared for all the jobs running on it.
If you want to run a job that deals with unbounded data, this job is not intended to end, ever. You want to be able to upgrade the job and redeploy the cluster with the new job, instead of dealing with resubmitting jobs, hence a job cluster feels more appropriate.

Now, let's continue with our adventure (using Flink 1.9.2).

Kubernetes: Job or Deployment?

Flink, in their official example advices to use a kubernetes job for the job-manager. This makes no sense IMHO as you want your job manager to be a long running application and automatically restart and continue from where it stopped if the pod gets deleted.

This is why I decided to change the job to a deployment.

Probes

Probes is a useful feature in kubernetes that helps us makes sure the application is running.

With Flink it's pretty easy to configure a liveness probe by accessing the Flink dashboard ui.

You can find that in the jobmanager-ha.yaml yaml.

Flink Configuration

Another thing I didn't like was the fact configuration is passed to flink via the CLI in the k8s container arguments.
This is why I created a configmap and use it to set Flink's configuration, both for the job and task managers.
You can find the definition in the flink-configuration-ha.yaml file.

Web UI

I added a rest service to be able to access Flink's web ui.
You can find the definition in the jobmanager-rest-service.yaml file.

You can find my fully working example here.

Don't forget to remove the imagePullPolicy: Never and set a real image name in the job manager and task manager yamls to run it in a non-minikube environment.

In the next blog post I cover the details of deploying a highly available Flink job cluster on k8s without ZooKeeper, using a file-based high availability implementation.

Enjoy your Flink adventures!

Understanding how to access an API you just created in AWS should be easy, right?

I spent a significant amount of time figuring out how to deploy a private API Gateway that would be accessible from outside of the VPC in which it resides.

Let's start with some background:
You can deploy an API Gateway using one of three modes: Regional, Edge and Private. If you want your API to be open to the internet, you'll probably choose Regional or Edge. If it's an API accessed by internal clients from within your VPC, you will probably choose Private.

All clients inside the gateway's VPC should be able to access it without a problem using the invoke url (you can find the invoke url at the top of the stage page).

It gets complicated when you want clients sitting in another VPC or your colleagues from the office network to be able to access your private API Gateway.
In this case, they won't be able to access it using the invoke url, as this DNS works only inside the VPC.

What should you do to solve it?
You need to create a VPC Endpoint for the VPC your API Gateway resides in, by going to Endpoints in the VPC management page.
Then creating the endpoint, choose the com.amazonaws.us-east-1.execute-api service, and your VPC. Tick the Enable Private DNS Name checkbox and create or use an existing security group to restrict the access to the VPC endpoint to your internal IPs only (your office internal subnet for example).

Then, to access the API Gateway, you must send a special header with every request, stating the API you want to access. The reason for this being the fact that you have 1 VPC Endpoint, and potentially more than one API Gateway inside this VPC.
The header name is x-apigw-api-id and the value should be the unique ID of your API Gateway.

An example request should look like this:

curl -X GET https://{VPC_ENDPOINT_DNS_NAME}/{STAGE_NAME}/{ENDPOINT_NAME} -H’x-apigw-api-id:{API_GW_ID}

or using Python:

requests.get('https://{VPC_ENDPOINT_DNS_NAME}/{STAGE_NAME}/{ENDPOINT_NAME}', headers={'x-apigw-api-id': '{API_GW_ID}'})

Unmuting doesn't work in Xfce?
I had this problem too after installing Debian Testing (Jessie) which now comes with Xfce by default, but you may encounter it on any system running Xfce DE.

Every time I muted the audio using the mute key on my keyboard, everything worked find - the audio was indeed muted. But when I tried to unmute using the same key, the audio indicator showed the unmuted state but audio was still muted.
After some investigation I understood that the only way to unmute my system was entering the command:

alsamixer

in the terminal and then press 'M' to toggle the mute state.
Apparently, the Audio Mixer was muting PulseAudio AND Alsa, but unmuting ONLY Alsa.

If you run the following command, it will list all the properties under xfce4-mixer:

xfconf-query -lc xfce4-mixer

In my case, what I saw is only one mixer under /sound-cards/ entry. It was Alsa, PulseAudio just wasn't there.
Also, in the audio mixer I saw only the Alsa entry and couldn't even control the PulseAudio mixer as you can see here:

From what I understood, xfce4-mixer (the Audio Mixer) was missing an optional dependency to be able to work with PulseAudio.
This dependency is: gstreamer0.10-pulseaudio.
After installing the above dependency, using (on debian):
sudo apt-get install gstreamer0.10-pulseaudio
the Audio Mixer looked a lot better:

But, running:

xfconf-query -c xfce4-mixer -p /active-card

to query the active card used by xfce4-mixer still returned "HDANVidiaAlsamixer".
So the last thing I needed to do is change the "active card" in xfce4-mixer too by running:

xfconf-query -c xfce4-mixer -p /active-card -s PlaybackBuiltinAudioAnalogStereoPulseAudioMixer

After that, toggling mute finally worked.