Logstash not sending data to elastic search

Elastic Stack Logstash
1

Trying to send data from logstash to an ES index, but the index is not created when logstash is run.

Tried logginf the input lines and there is rows logged in the logstash log, but data not sent to ES index

jdbc {
    type => "csvideo"
    jdbc_connection_string => "jdbc:postgresql://${DB_HOST}:${DB_PORT}/${DB_NAME}?useCursorFetch=true"
    jdbc_user => "${DB_USER}"
    jdbc_password => "${DB_PASSWORD}"
    schedule => "* * * * *"
    jdbc_driver_class => "org.postgresql.Driver"
    statement => "with tags as ( SELECT csvideo_id, string_agg(DISTINCT tag_name, ' >< ') AS tags, count(*) AS tag_count
      FROM csvideo_tag
      GROUP BY csvideo_id),
medias as ( SELECT csvideo_id, media_org, media_thumb
      FROM csvideo_media
      WHERE seq = 1 
      GROUP BY csvideo_id, media_org, media_thumb),
products as ( SELECT csvideo_id, string_agg(item_cd::varchar, ' >< ') AS item_codes, count(*) AS product_count
      FROM csvideo_product
      GROUP BY csvideo_id)
select csvideos.csvideo_id,
      'CS Video' as type,
      csvideos.sb_id,
      csvideos.user_code,
      csvideos.comment,
      csvideos.shop_code,
      csvideos.style_height,
      csvideos.styling_category,
      csvideos.scheduled_at,
      csvideos.created_at,
      csvideos.published_at,
      csvideos.last_updated,
      csvideos.stm_comments,
      csvideos.status,
      csvideos.is_deleted,
      csvideos.created_by,
      csvideos.favorite_count,
      csvideos.can_display_csvideo_on_st,
      br.shop_no AS shop_number,
      profile.label_code, 
      profile.label_name, 
      profile.shop_name,
      profile.user_nickname, 
      tags.tags,
      tags.tag_count,
      medias.media_org,
      medias.media_thumb,
      products.item_codes,
      products.product_count
      from
      csvideo_master csvideos
      LEFT JOIN staff_profile profile ON csvideos.sb_id = profile.sb_id
	  LEFT JOIN brand br ON br.label_code = profile.label_code
      left join tags
      on csvideos.csvideo_id = tags.csvideo_id
      left join medias on csvideos.csvideo_id = medias.csvideo_id
      left join products on csvideos.csvideo_id = products.csvideo_id
	  WHERE
      csvideos.last_updated > :sql_last_value::TIMESTAMP
      AND csvideos.status <> 'draft'
    order by csvideos.last_updated ASC"
    use_column_value => true
    tracking_column => "last_updated"
    tracking_column_type => "timestamp"
    last_run_metadata_path => "/usr/share/logstash/logstash_jdbc_last_run_csvideo"
    jdbc_paging_enabled => true
    jdbc_page_size => 50000
    jdbc_fetch_size => 50000
  }

  if [type] == "csvideo" {
    elasticsearch {
      hosts => ["${ELASTIC_HOST}"]
      ssl => true
      index => "${ELASTIC_CSVIDEO_INDEX}"
      document_id => "%{csvideo_id}"
      user => "${ELASTIC_USER_NAME}"
      password => "${ELASTIC_PASSWORD}"
      ilm_enabled => false
    }
  }

Data gets send to ES index when
if [type] == "csvideo"
is not specified in the output.

I tried executing the query in a db client and it returns many rows of data

I think there could be something wrong with how the query is aligned, when I give a simple select * query as statement it is reflected in ES index. Cant figure out the issue with the query.

Any help is appreciated.

2

As far as I know, the jdbc input does not have a type option. This should result in an error when logstash starts.

3

No that is not the issue, this is working for other cases

4

You are correct, it is one of the "common options" supported by the base input plugin class that other inputs extend. Sorry about that.

1 Like
5

Memory is fuzzy but I remember writing a long query with multiple joins and at some point they just stopped working. Had to convert the entire SQL query over to a CTE instead of joins to make it work. Still don't know what the root cause is for this but I think it's similar to what I've seen.

I would try 3 things in this order.

  1. Try putting the SQL statement in a file and use statement_filepath.
  2. Depending on your DB type you might be able to execute this as a stored procedure.
  3. Convert query to CTE.
6

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.